diff --git a/backend/app/api/cashflow_structure.py b/backend/app/api/cashflow_structure.py index 5f950dd3..3255cb18 100644 --- a/backend/app/api/cashflow_structure.py +++ b/backend/app/api/cashflow_structure.py @@ -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, } diff --git a/backend/app/utils/caliber_units.py b/backend/app/utils/caliber_units.py new file mode 100644 index 00000000..5d012440 --- /dev/null +++ b/backend/app/utils/caliber_units.py @@ -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()) diff --git a/backend/scripts/alert_generator.py b/backend/scripts/alert_generator.py index 0f0850f5..92fecb0d 100644 --- a/backend/scripts/alert_generator.py +++ b/backend/scripts/alert_generator.py @@ -6,8 +6,10 @@ import sys import os +import re import json import logging +from typing import Optional from datetime import datetime, timedelta sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) @@ -15,9 +17,36 @@ sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) from app.database import get_session_local from app.models import KPIDefinition, KPIValue, KPIAlert, ActionPlan from app.utils.threshold_parser import eval_threshold, LEVEL_PRIORITY +# 裁定2a（2026-09-17 财务口径）：比率类指标统一「百分比(%)」口径（取数×100）， +# 未核对/疑似单位错配的项降级待核（不落预警、不推送） +from app.utils.caliber_units import resolve_caliber logger = logging.getLogger("cma.alert_gen") +#: 月期间正则 —— 只认 YYYY-MM（排除 2026-H1/H2、2025-FY 等伪期间；裁定2b） +PERIOD_REGEX = r"^[0-9]{4}-[0-9]{2}$" +_PERIOD_RE = re.compile(PERIOD_REGEX) + + +def resolve_latest_period(db_session, kpi_id=None): + """「最近可用期间」（裁定2b）：kpi_values 内最新 YYYY-MM，排除 H1/H2 伪期间。 + + 给定 kpi_id → 返回**该 KPI 自己**的最近可用期间：期间口径不硬编码当月， + 也不被别人（他账套/演示账套）更新的新期间带偏——实证：全局 MAX(period)=2026-09 + 但那是演示账套 e6 的 8 条数据，酣客(e1) 的最新可用期间是 2026-08。 + + 实现说明：伪期间过滤在 **Python 侧**做（`SELECT DISTINCT period` 只有十几行）， + 不依赖 MySQL 的 `REGEXP` —— SQLite 测试库不注册 REGEXP，用 SQL 侧正则会让 + 单测与生产走两套语义。 + """ + q = db_session.query(KPIValue.period).filter(KPIValue.actual_value.isnot(None)) + if kpi_id is not None: + q = q.filter(KPIValue.kpi_id == kpi_id) + for (p,) in q.distinct().order_by(KPIValue.period.desc()).all(): + if _PERIOD_RE.fullmatch(str(p or "")): + return p + return None + def _parse_threshold(expr: str) -> tuple: """[兼容保留] 统一委托 app.utils.threshold_parser.parse_threshold""" @@ -56,10 +85,23 @@ def _check_threshold(actual: float, threshold_expr: str, level: str) -> tuple: return True, msg -def run_alert_check(db_session, period: str = None) -> int: - """检查所有KPI的实际值是否触发预警，返回新生成的预警数""" - if period is None: - period = datetime.now().strftime("%Y-%m") +def run_alert_check(db_session, period: Optional[str] = None, dry_run: bool = False, + stats: Optional[dict] = None) -> int: + """检查所有KPI的实际值是否触发预警，返回（将）新生成的预警数 + + 2026-09-17 财务口径裁定 2a/2b： + · **期间（2b）**：period=None → 每个 KPI 回落「最近可用期间」（自己的最新 YYYY-MM， + 排除 H1/H2 伪期间），不再硬编码当月（当月无值 → 永久 0 新预警）； + · **单位（2a）**：判定前经 resolve_caliber() 归一到「百分比(%)」口径（取数×100， + 与阈值 150/80 同量纲）；**未核对/疑似单位错配**的项降级待核 → 不落预警、不推送； + · dry_run=True → 只计算不写库（抽查复核与放行前证据用）。 + """ + st = stats if stats is not None else {} + for k, v in (("period_source", "fixed" if period else "per_kpi_latest"), ("period_fixed", period), + ("periods_used", {}), ("normalized", 0), ("normalized_kpis", []), + ("dimension_suspect", 0), ("suspect_kpis", []), ("no_period", 0), + ("no_value", 0), ("planned", []), ("skipped_existing", 0)): + st.setdefault(k, v) kpis = db_session.query(KPIDefinition).filter( KPIDefinition.status == "active" @@ -76,16 +118,39 @@ def run_alert_check(db_session, period: str = None) -> int: if not any(thr.values()): continue - # 获取该KPI当前期间的最新值 + # 期间口径（裁定2b）：显式指定 → 固定期间；缺省 → 该 KPI 的最近可用期间 + p = period or resolve_latest_period(db_session, kpi.id) + if not p: + st["no_period"] += 1 + logger.warning(f" 跳过（无可用期间）: {kpi.kpi_name} 无任何 YYYY-MM 实测值") + continue + + # 获取该KPI该期间的最新值 latest = db_session.query(KPIValue).filter( KPIValue.kpi_id == kpi.id, - KPIValue.period == period, + KPIValue.period == p, ).order_by(KPIValue.calculated_at.desc()).first() if not latest or latest.actual_value is None: + st["no_value"] += 1 continue - actual = latest.actual_value + # 单位口径（裁定2a）：取数归一 + 疑似错配降级门 + cal = resolve_caliber(latest.actual_value, kpi) + if cal["status"] == "dimension_suspect": + st["dimension_suspect"] += 1 + st["suspect_kpis"].append("%s[%s] actual=%s unit=%s" % ( + kpi.kpi_code, p, latest.actual_value, kpi.unit)) + logger.warning(" [降级待核·疑似单位错配→不落预警不推送] %s[%s] 实测=%s 单位=%s ← %s" % ( + kpi.kpi_name, p, latest.actual_value, kpi.unit, cal["reason"])) + continue + + actual = cal["value"] + if cal["status"] == "normalized": + st["normalized"] += 1 + st["normalized_kpis"].append("%s[%s] %s → %s（×%s）" % ( + kpi.kpi_code, p, latest.actual_value, actual, cal["factor"])) + st["periods_used"][p] = st["periods_used"].get(p, 0) + 1 # 判定顺序：红灯优先 → 绿灯 → 黄灯（最严重者优先，消除宽黄灯遮蔽红灯） # 2026-09-11 修复「黄灯遮蔽红灯」：原顺序 绿→黄→红命中即止，宽黄灯(如 Y='>76') @@ -101,6 +166,8 @@ def run_alert_check(db_session, period: str = None) -> int: if level == "green": # 正常状态：不生成待处理预警（如有旧的red/yellow预警自动解除） break + if cal.get("note"): + msg = "%s｜%s" % (msg, cal["note"]) # 检查是否已有该期间该KPI同级别的预警 existing = db_session.query(KPIAlert).filter( KPIAlert.kpi_id == kpi.id, @@ -110,6 +177,16 @@ def run_alert_check(db_session, period: str = None) -> int: if existing: # 已有预警，跳过 + st["skipped_existing"] += 1 + break + + st["planned"].append({ + "kpi": "%s/%s" % (kpi.kpi_code, kpi.kpi_name), "period": p, + "actual": actual, "raw_actual": latest.actual_value, "factor": cal["factor"], + "level": level, "message": msg, + }) + if dry_run: + new_count += 1 break # 创建新预警 @@ -118,7 +195,7 @@ def run_alert_check(db_session, period: str = None) -> int: kpi_id=kpi.id, kpi_value_id=latest.id, alert_level=level, - alert_message=f"{kpi.kpi_name}[{period}] {msg}", + alert_message=f"{kpi.kpi_name}[{p}] {msg}", status="pending", ) db_session.add(alert) @@ -160,12 +237,15 @@ def run_alert_check(db_session, period: str = None) -> int: KPIAlert.alert_level == "red", KPIAlert.status.in_(["pending", "processing"]), ).first() - if existing_red: + if existing_red and not dry_run: existing_red.status = "resolved" existing_red.resolution = "自动解除: 当前值不再触发红色阈值" existing_red.resolved_at = datetime.now() logger.info(f" 自动解除预警 [red] {kpi.kpi_name}") + if dry_run: + db_session.rollback() # 抽查复核/放行前证据：确保零写入 + return new_count db_session.commit() return new_count @@ -205,22 +285,30 @@ def run_rule_engine_checks(db_session) -> dict: return stats -def generate_and_push(db_session) -> dict: - """生成预警并推送，返回统计""" - period = datetime.now().strftime("%Y-%m") +def generate_and_push(db_session, period: Optional[str] = None, dry_run: bool = False) -> dict: + """生成预警并推送，返回统计 - # 0. Bug1修复: 先清理历史green预警 - cleaned = cleanup_green_alerts(db_session) + 2026-09-17 财务口径裁定 2a/2b： + · **period=None** → 判定期间 = 各 KPI「最近可用期间」（裁定2b，不再硬编码当月）； + 显式给出 period → 固定该期间（抽查复核用）； + · **dry_run=True** → 零写入、零推送：返回将生成的预警清单与口径证据（放行前抽查复核）。 + """ + cur_month = datetime.now().strftime("%Y-%m") + st = {} + logger.info("开始预警检查（当月=%s；判定期间=%s）..." % ( + cur_month, ("固定 %s" % period) if period else "各 KPI 最近可用期间（裁定2b）")) - # 1. 生成预警 - logger.info(f"开始预警检查 ({period})...") - new_count = run_alert_check(db_session, period) - logger.info(f"预警检查完成: 新增 {new_count} 条") + # 0. Bug1修复: 先清理历史green预警（dry-run 不写） + cleaned = 0 if dry_run else cleanup_green_alerts(db_session) - # 1a. 覆盖度自检（2026-09-16 新增，任务 finance-20260916-cma-health-gaps-3items · 项3）： - # run_alert_check 的期间口径 = 当月（period=current month），若当月尚无任何 KPIValue， - # 则「0 条新预警」并不代表"确无阈值触发"，而是"本次检查未覆盖任何数据"—— - # 该区别原实现不可见（日志与汇总行都只说 0 条），导致连续多日 0 新增无法归因。 + # 1. 生成预警（期间口径 + 单位口径均在此落地） + new_count = run_alert_check(db_session, period=period, dry_run=dry_run, stats=st) + logger.info(f"预警检查完成: {'将新增' if dry_run else '新增'} {new_count} 条") + + # 1a. 覆盖度自检（2026-09-16 项3；2026-09-17 裁定2b 过渡期保留并扩展「期间回落」明细）： + # 若**当月**无任何 KPIValue，「0 条新预警」并不代表"确无阈值触发"，而是"当月无数据"； + # 裁定2b 起改为按各 KPI 最近可用期间判定，故此处显式记录回落后的期间分布。 + coverage = {"current_month": cur_month, "periods_used": st.get("periods_used", {})} try: kpis_with_values = 0 active_kpis = db_session.query(KPIDefinition).filter(KPIDefinition.status == "active").all() @@ -228,33 +316,30 @@ def generate_and_push(db_session) -> dict: if not any((k.threshold_red, k.threshold_yellow, k.threshold_green)): continue hit = db_session.query(KPIValue).filter( - KPIValue.kpi_id == k.id, KPIValue.period == period, + KPIValue.kpi_id == k.id, KPIValue.period == cur_month, KPIValue.actual_value.isnot(None), ).first() if hit: kpis_with_values += 1 - coverage = {"period": period, "kpis_with_threshold_and_value": kpis_with_values} + coverage["kpis_with_threshold_and_value"] = kpis_with_values + coverage["period"] = cur_month if kpis_with_values == 0: - latest_period = db_session.query(KPIValue.period).filter( - KPIValue.actual_value.isnot(None), - KPIValue.period.op("REGEXP")("^[0-9]{4}-[0-9]{2}$"), # 只取 YYYY-MM 月期间（排除 2026-H1/H2 伪期间） - ).order_by(KPIValue.period.desc()).first() + coverage["latest_period_in_db"] = resolve_latest_period(db_session) logger.warning( - f"覆盖度告警: 当月({period})无任何 KPI 实际值 → 本次预警检查未覆盖任何 KPI，" - f"「0 条新预警」= 无数据可判，非「确无阈值触发」。库内最新期间=" - f"{latest_period[0] if latest_period else 'N/A'}" - ) - coverage["latest_period_in_db"] = latest_period[0] if latest_period else None + "覆盖度告警: 当月(%s)无任何 KPI 实际值 → 本次按「最近可用期间」判定（期间分布 %s）；" + "「0 条新预警」= 当月无数据可判，非「确无阈值触发」。库内最新月期间=%s" % ( + cur_month, st.get("periods_used"), coverage["latest_period_in_db"])) except Exception as e: logger.error(f"覆盖度自检失败: {e}") coverage = {"error": str(e)} - # 1b. 规则引擎检查（预测偏差 / 客户集中度，P2 2026-09-10） - rule_stats = run_rule_engine_checks(db_session) + # 1b. 规则引擎检查（预测偏差 / 客户集中度，P2 2026-09-10；dry-run 不写） + rule_stats = {"forecast_deviation": 0, "customer_concentration": 0} if dry_run \ + else run_rule_engine_checks(db_session) - # 2. 推送 + # 2. 推送（dry-run 零推送） pushed = 0 - if new_count > 0: + if new_count > 0 and not dry_run: try: from app.utils.notifier import push_pending_alerts pushed = push_pending_alerts(db_session) @@ -262,24 +347,51 @@ def generate_and_push(db_session) -> dict: except Exception as e: logger.error(f"推送失败: {e}") - return {"period": period, "new_alerts": new_count, "pushed": pushed, + return {"period": cur_month, "dry_run": dry_run, + "period_source": st.get("period_source"), "period_fixed": st.get("period_fixed"), + "periods_used": st.get("periods_used", {}), + "new_alerts": new_count, "pushed": pushed, "cleaned_green": cleaned, "rule_engine": rule_stats, - "coverage": coverage} + "coverage": coverage, + # 裁定2a 口径证据 + "normalized": st.get("normalized", 0), "normalized_kpis": st.get("normalized_kpis", []), + "dimension_suspect": st.get("dimension_suspect", 0), "suspect_kpis": st.get("suspect_kpis", []), + # 期间/取值覆盖明细 + "no_period": st.get("no_period", 0), "no_value": st.get("no_value", 0), + "skipped_existing": st.get("skipped_existing", 0), "planned": st.get("planned", [])} def coverage_note(result: dict) -> str: - """汇总行追加提示：当月无 KPI 值时显式说明「0 条 = 无数据可判」而非「确无触发」 + """汇总行追加提示：期间口径回落后显式说明「0 条 = 无数据可判」而非「确无触发」 - 2026-09-16 新增（项3）。独立成函数便于回查（无需触发真实写路径即可断言文案）。 + 2026-09-16 新增（项3）；2026-09-17 裁定2b 起追加「期间回落」明细。 """ cov = result.get("coverage") or {} + parts = [] + used = result.get("periods_used") or cov.get("periods_used") or {} + if used and cov.get("current_month") not in used: + parts.append(" | 期间回落(裁定2b): 当月(%s)无 KPI 值 → 按各 KPI 最近可用期间判定 %s" + % (cov.get("current_month"), used)) if cov.get("kpis_with_threshold_and_value") == 0: - return (f" | ⚠️ 当月({result['period']})无KPI值: 本次检查未覆盖任何KPI" - f"（库内最新月期间 {cov.get('latest_period_in_db')}）→ 0 条不等于确无触发") - return "" + parts.append(" | ⚠️ 当月(%s)无KPI值: 本次检查按最近可用期间判定（库内最新月期间 %s）→ 0 条不等于确无触发" + % (cov.get("current_month"), cov.get("latest_period_in_db"))) + if result.get("dimension_suspect"): + parts.append(" | ⛔ 降级待核(疑似单位错配，不推送) %s 项: %s" + % (result["dimension_suspect"], (result.get("suspect_kpis") or [])[:3])) + if result.get("normalized"): + parts.append(" | 口径归一 ×100 %s 项" % result["normalized"]) + return "".join(parts) if __name__ == "__main__": + import argparse + + ap = argparse.ArgumentParser(description="预警生成与推送（2026-09-17 口径裁定 2a/2b）") + ap.add_argument("--dry-run", action="store_true", help="只计算不写库不推送（抽查复核/放行前证据）") + ap.add_argument("--period", default=None, help="固定判定期间 YYYY-MM（缺省=各 KPI 最近可用期间，裁定2b）") + ap.add_argument("--json", action="store_true", help="JSON 输出完整证据（含将生成的预警清单）") + args = ap.parse_args() + logging.basicConfig( level=logging.INFO, format="%(asctime)s [%(levelname)s] %(name)s: %(message)s", @@ -287,8 +399,13 @@ if __name__ == "__main__": db = get_session_local()() try: - result = generate_and_push(db) - print(f"预警检查: {result['new_alerts']}条新预警 / {result['pushed']}条已推送" - f" / 规则引擎{result['rule_engine']}{coverage_note(result)}") + result = generate_and_push(db, period=args.period, dry_run=args.dry_run) + if args.json: + print(json.dumps(result, ensure_ascii=False, indent=1, default=str)) + else: + print(f"预警检查{'（dry-run 零写入零推送）' if args.dry_run else ''}: " + f"{result['new_alerts']}条新预警 / {result['pushed']}条已推送" + f" / 规则引擎{result['rule_engine']}{coverage_note(result)}") finally: db.close() + diff --git a/backend/scripts/import_cashflow_statements.py b/backend/scripts/import_cashflow_statements.py index 7f0c7340..aeea5706 100644 --- a/backend/scripts/import_cashflow_statements.py +++ b/backend/scripts/import_cashflow_statements.py @@ -15,8 +15,9 @@ - **NA（无发生额）标注（本版本新增，裁定 finance-20260916-cma-caliber-ruling §2.1）**： 某活动**无任何项目行**时净额记 0.00（三净额列为 NOT NULL，表本身表达不了 NA）， 同时在 `remark` 写入**固定短语**「<活动>活动本期无发生额（无项目行）」， - 供判定层用 `parse_na_activities(remark)` 取回 → 判定时该维度**不参与**（降维）， - 避免把「无发生额」与「收付相抵的 0」混为一谈（后者才参与判定）。 + 供判定层用 `parse_na_activities(remark)` 取回 → **仅作标注与机读入口**； + 口径裁定（2026-09-17）已定：该维按 **0 如实参与判定**（八类判定的**二维映射**， + 经营(±)×筹资(±) 出 2 个候选类型），**不再**因 NA/0 整体判「未判定」。 - source='official'（官方报表口径）；期初/期末现金 留空（主表采集缺陷，见报告 §项1-②） - 同 (entity_id, period) 重复执行 = 覆盖更新（幂等） @@ -52,8 +53,10 @@ _NA_RE = re.compile(r"(经营活动|投资活动|筹资活动)本期无发生额 def parse_na_activities(remark): """从 cashflow_statements.remark 取回 NA（无发生额）活动集合。 - 判定层入口：NA 维度**不参与**八类判定（降维），其 0.00 是"表表达不了 NA"的占位， - 不是"收付相抵的 0"。同时提供 `na_flags(remark)` 返回 {活动: True/False} 供逐维判断。 + 判定层入口（**只作标注/机读**）：NA 维度的 0.00 是"表表达不了 NA"的占位， + 不等于"收付相抵的 0"；但按 2026-09-17 口径裁定，两者**都按 0 参与判定** + （投资活动=0 → 二维映射出 2 个候选类型），NA 短语仅用于页面/报告显式标注。 + `na_flags(remark)` 返回 {活动: True/False} 供逐维判断与文案生成。 """ if not remark: return set() @@ -117,7 +120,7 @@ def collect(period_filter=None): na_phrases = [NA_PHRASE_TMPL.format(cls=c) for c in missing] head = "源: 智享通现金流量统计表 %s（单位元→万元 /10000）；项目行 %d 条" % ( os.path.basename(src), len(rows)) - remark = (head + "；NA(无发生额，不参与判定): " + "、".join(na_phrases)) if na_phrases \ + remark = (head + "；NA(无发生额，按0参与判定): " + "、".join(na_phrases)) if na_phrases \ else (head + "；三类活动均有项目行（无 NA）") payloads.append({ "period": period, @@ -145,7 +148,7 @@ def _self_test(): ([], {"经营活动": False, "投资活动": False, "筹资活动": False}), ): phrases = [NA_PHRASE_TMPL.format(cls=c) for c in missing] - remark = ("源: X；项目行 3 条；NA(无发生额，不参与判定): " + "、".join(phrases)) if phrases \ + remark = ("源: X；项目行 3 条；NA(无发生额，按0参与判定): " + "、".join(phrases)) if phrases \ else "源: X；项目行 3 条；三类活动均有项目行（无 NA）" assert na_flags(remark) == expect, (missing, remark) assert parse_na_activities(remark) == set(missing) @@ -198,12 +201,23 @@ def main(): continue n = p["nets_wan"] cls = classify_enterprise(n["经营活动"], n["投资活动"], n["筹资活动"]) + # 裁定1（2026-09-17）：投资活动=0 → dimension="2D"（type_no=None，候选 2 个） + # ⇒ 打印层不得假定 classified ⇒ type_no 非空（原实现此处 TypeError 崩掉整轮 dry-run） + if cls.get("classified"): + if cls.get("dimension") == "2D": + verdict = "二维判定(%s) → 候选 %s" % ( + cls.get("dimension_label") or "投资活动=0", + " / ".join("类型%d %s" % (c["type_no"], c["name"]) + for c in cls.get("candidates") or [])) + else: + verdict = "类型%d %s" % (cls["type_no"], cls["name"]) + else: + verdict = "未判定: " + cls.get("reason", "")[:28] print(" %-8s O=%8.2f I=%8.2f F=%8.2f 万 | 判定=%s | 源=%s" % (p["period"], n["经营活动"], n["投资活动"], n["筹资活动"], - ("类型%d %s" % (cls["type_no"], cls["name"])) if cls.get("classified") - else ("未判定: " + cls.get("reason", "")[:28]), - os.path.basename(p["src"]))) - print(" NA(无发生额，不参与判定)=%s" % ("、".join(p["na_classes"]) or "无")) + verdict, os.path.basename(p["src"]))) + print(" NA(无发生额)=%s；投资栏标注=%s" % ( + "、".join(p["na_classes"]) or "无", cls.get("investing_note") or "-")) if not args.apply: continue rec = db.query(CashFlowStructureValue).filter( diff --git a/backend/scripts/kpi_unit_crosscheck.py b/backend/scripts/kpi_unit_crosscheck.py new file mode 100644 index 00000000..0d0ac594 --- /dev/null +++ b/backend/scripts/kpi_unit_crosscheck.py @@ -0,0 +1,99 @@ +#!/usr/bin/env python3 +# -*- coding: utf-8 -*- +"""比率类 KPI 单位对照表（只读）—— 财务口径裁定 裁定2a 的产出器 + +用途：逐个比率类 KPI 核对「定义 / 声明单位 / 阈值量纲 / 实测值量纲」，出对照表。 + 判定侧的事实源是 `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）===") + 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()) diff --git a/backend/scripts/migrate_kpi_threshold_unit_20260916.py b/backend/scripts/migrate_kpi_threshold_unit_20260916.py index 850d1449..9874a127 100644 --- a/backend/scripts/migrate_kpi_threshold_unit_20260916.py +++ b/backend/scripts/migrate_kpi_threshold_unit_20260916.py @@ -2,7 +2,17 @@ # -*- coding: utf-8 -*- """阈值单位口径修正 + 量纲守卫装配（迁移脚本·准备态） -依据：财务口径裁定 finance-20260916-cma-caliber-ruling.md v1.0 §四（裁定③）+ §3.3/§3.4 +⚠️ **已被取代，不得执行（2026-09-17）**： + 本脚本基于 2026-09-16 的口径草案（**把阈值改成「倍」写法** >=2.0/>=1.0/<1.0）。 + 2026-09-17 财务**正式裁定**（`wecom-finance/reports/CMA_口径裁定_20260917.md` 裁定2a） + 采用**相反方向**：比率类统一「百分比(%)」口径、**阈值保持不变**（150/80）， + 在**取数层** ×100 归一（实现：`app/utils/caliber_units.py` + `scripts/alert_generator.py`）。 + ⇒ 执行本脚本会把口径反向切回「倍」，与现行裁定冲突；假红灯 586/587 在新口径下 + **是合法红灯**（流动比率 1.29=129% < 150，与「流动性偏紧」一致），**不得作废**。 + 保留本文件仅为审计链（历史草案）——如未来另有阈值单位新裁定，须先改 `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 按「口径错误」作废并重算 diff --git a/backend/tests/test_caliber_rulings_20260917.py b/backend/tests/test_caliber_rulings_20260917.py new file mode 100644 index 00000000..58654ad2 --- /dev/null +++ b/backend/tests/test_caliber_rulings_20260917.py @@ -0,0 +1,196 @@ +# -*- coding: utf-8 -*- +"""财务口径裁定 2026-09-17 落地测试（裁定1 / 2a / 2b） + +依据：`wecom-finance/reports/CMA_口径裁定_20260917.md` + · 裁定1：现金流「投资活动净额=0」= 本期无投资活动，**按 0 如实参与判定**（二维映射 + 标注）； + · 裁定2a：比率类 KPI 统一「百分比(%)」口径（取数×100），未核对/疑似错配**不推送**； + · 裁定2b：预警期间由「当月硬编码」→「最近可用期间」（排除 2026-H1/H2 伪期间）。 +""" +import pytest +from sqlalchemy.orm import Session + +from tests.conftest import create_test_kpi +from app.api.cashflow_structure import classify_enterprise, INVESTING_ZERO_NOTE +from app.utils.caliber_units import resolve_caliber, RATIO_CALIBER_TABLE +from scripts.alert_generator import resolve_latest_period, run_alert_check +from app.models import KPIValue, KPIAlert + + +class _K: + """最小 KPI 载体（duck type）""" + + def __init__(self, kpi_code, unit, green=None, yellow=None, red=None, kpi_id=None): + self.id = kpi_id + self.kpi_code, self.unit = kpi_code, unit + self.threshold_green, self.threshold_yellow, self.threshold_red = green, yellow, red + + +# ───────────────────────── 裁定1：投资活动=0 ───────────────────────── + +class TestRuling1InvestingZero: + def test_investing_zero_2d_mapping(self): + """投资=0 + 经营(+)/筹资(-) → 二维映射候选 1/2，并标注「本期无投资活动」""" + c = classify_enterprise(14.45, 0.0, -3.2) + assert c["classified"] is True + assert c["dimension"] == "2D" + assert c["candidate_type_nos"] == [1, 2] + assert c["type_no"] is None + assert c["investing_note"] == INVESTING_ZERO_NOTE + assert c["o_sign"] == "+" and c["i_sign"] == "0" and c["f_sign"] == "-" + assert "二维" in c["reason"] and "类型1" in c["reason"] and "类型2" in c["reason"] + + def test_investing_zero_all_pairs(self): + expect = {(1, 1): [3, 4], (1, -1): [1, 2], (-1, 1): [6, 7], (-1, -1): [5, 8]} + for (o, f), cands in expect.items(): + c = classify_enterprise(o * 10.0, 0.0, f * 5.0) + assert c["classified"] is True and c["candidate_type_nos"] == cands, (o, f) + + def test_investing_zero_with_other_zero_is_undetermined(self): + """经营/筹资亦为 0 → 回退「未判定（无投资活动）」，不报错、不静默""" + c = classify_enterprise(0.0, 0.0, -3.0) + assert c["classified"] is False + assert "未判定" in c["reason"] and "无投资活动" in c["reason"] + assert c["investing_note"] == INVESTING_ZERO_NOTE + c2 = classify_enterprise(5.0, 0.0, 0.0) + assert c2["classified"] is False and "未判定" in c2["reason"] + + def test_three_dim_still_exact_type(self): + """三维齐备时判定保持原语义（无回归）""" + c = classify_enterprise(-10.0, 3.0, 12.0) + assert (c["classified"], c["dimension"], c["type_no"], c["name"]) == (True, "3D", 7, "输血生存型") + assert c["investing_note"] is None + + def test_missing_value_reason(self): + c = classify_enterprise(None, 0.0, 12.0) + assert c["classified"] is False and "缺失" in c["reason"] + + +# ───────────────────────── 裁定2a：单位口径 ───────────────────────── + +class TestRuling2aCaliber: + def test_current_ratio_normalized_to_percent(self): + cal = resolve_caliber(1.29, _K("F_CURRENT_RATIO", "%", ">=200", ">=150", "<150")) + assert cal["status"] == "normalized" and cal["factor"] == 100 + assert round(cal["value"], 2) == 129.0 + assert "口径归一" in cal["note"] + + def test_percent_class_not_multiplied(self): + """返利率/收入增长率等取数已是百分数 → 因子 1（再 ×100 会反向越界）""" + for code, val in (("F_REBATE_RATE", 48.44), ("F_REVENUE_GROWTH", -82.2), + ("F_DEBT_RATIO", 39.9), ("F_COST_RATIO", 30.0)): + cal = resolve_caliber(val, _K(code, "%", ">=10", ">=8", "<5")) + assert cal["status"] == "ok" and cal["factor"] == 1, code + assert cal["value"] == val + + def test_small_real_value_not_flagged(self): + """总资产周转率 0.008 次 vs 阈值 1.0 次 = 真实小值（非错配），不得误伤""" + cal = resolve_caliber(0.008, _K("F_ASSET_TURNOVER", "次", ">=1.0", ">=0.8", "<0.8")) + assert cal["status"] == "ok" and cal["factor"] == 1 + + def test_unverified_ratio_kpi_is_suspect(self): + """表外比率 KPI + 百分数写法阈值 → 疑似错配（降级待核，不推送）""" + cal = resolve_caliber(1.29, _K("X_UNKNOWN_RATIO", "%", ">=200", ">=150", "<150")) + assert cal["status"] == "dimension_suspect" + + def test_table_entries_have_evidence(self): + assert RATIO_CALIBER_TABLE + for code, row in RATIO_CALIBER_TABLE.items(): + assert row["norm_factor"] in (1, 100), code + assert row["verified"] is True, code + assert row["evidence"] and row["caliber"], code + + +# ───────────────────────── 裁定2b：期间回落 ───────────────────────── + +class TestRuling2bPeriod: + def _mk(self, db, code, period, value, **kw): + kpi = create_test_kpi(db, kpi_code=code, kpi_name=code, unit=kw.pop("unit", "%"), + threshold_red=kw.pop("red", "<150"), **kw) + db.add(KPIValue(entity_id=1, kpi_id=kpi.id, period=period, actual_value=value)) + db.commit() + return kpi + + def test_pseudo_period_excluded(self, db: Session): + """H1/H2/FY 伪期间必须被排除（否则期间回落会取到 2026-H2）""" + kpi = self._mk(db, "T_PERIOD", "2026-H2", 1.0) + db.add(KPIValue(entity_id=1, kpi_id=kpi.id, period="2026-08", actual_value=200.0)) + db.commit() + assert resolve_latest_period(db, kpi.id) == "2026-08" + assert resolve_latest_period(db) == "2026-08" + + def test_no_value_returns_none(self, db: Session): + kpi = create_test_kpi(db, kpi_code="T_EMPTY", kpi_name="空", unit="%", threshold_red="<1") + assert resolve_latest_period(db, kpi.id) is None + + def test_run_alert_check_uses_kpi_latest_period(self, db: Session): + """当前月无值 → 判定期间回落到该 KPI 最近可用期间（不再 0 预警）""" + kpi = self._mk(db, "T_FALLBACK", "2026-08", 14.45, red="<58.4") + st = {} + n = run_alert_check(db, period=None, dry_run=True, stats=st) + assert n == 1 + assert st["period_source"] == "per_kpi_latest" + assert st["periods_used"] == {"2026-08": 1} + assert st["planned"][0]["kpi"].startswith("T_FALLBACK") and st["planned"][0]["level"] == "red" + + +# ───────────────────────── 2a×2b 合流：降级不推送 + 零写入 ───────────────────────── + +class TestGateAndDryRun: + def test_suspect_kpi_not_alerted(self, db: Session): + """疑似单位错配的 KPI：dry-run 下 0 预警 + 计入 dimension_suspect（不落库不推送）""" + kpi = create_test_kpi(db, kpi_code="X_UNKNOWN_RATIO", kpi_name="未知比率", unit="%", + threshold_green=">=200", threshold_yellow=">=150", threshold_red="<150") + db.add(KPIValue(entity_id=1, kpi_id=kpi.id, period="2026-08", actual_value=1.29)) + db.commit() + st = {} + n = run_alert_check(db, period="2026-08", dry_run=True, stats=st) + assert n == 0, "疑似错配不得生成预警" + assert st["dimension_suspect"] == 1 and st["suspect_kpis"] + assert db.query(KPIAlert).count() == 0 + + def test_listed_ratio_kpi_alert_uses_normalized_value(self, db: Session): + """表内比率 KPI：1.29 → 129% 参与判定，且消息里带口径归一说明""" + kpi = create_test_kpi(db, kpi_code="F_CURRENT_RATIO", kpi_name="流动比率", unit="%", + threshold_green=">=200", threshold_yellow=">=150", threshold_red="<150") + db.add(KPIValue(entity_id=1, kpi_id=kpi.id, period="2026-08", actual_value=1.29)) + db.commit() + st = {} + n = run_alert_check(db, period="2026-08", dry_run=True, stats=st) + assert n == 1 and st["planned"][0]["level"] == "red" + assert st["planned"][0]["actual"] == 129.0 and st["planned"][0]["factor"] == 100 + assert "口径归一" in st["planned"][0]["message"] + assert st["normalized"] == 1 + + def test_dry_run_writes_nothing(self, db: Session): + kpi = create_test_kpi(db, kpi_code="T_DRY", kpi_name="dry", unit="万元", + threshold_green=">=97", threshold_yellow=">=78", threshold_red="<58") + db.add(KPIValue(entity_id=1, kpi_id=kpi.id, period="2026-08", actual_value=14.45)) + db.commit() + before = db.query(KPIAlert).count() + n_dry = run_alert_check(db, period=None, dry_run=True, stats={}) + assert n_dry == 1 and db.query(KPIAlert).count() == before, "dry-run 必须零写入" + n_real = run_alert_check(db, period=None, dry_run=False, stats={}) + assert n_real == 1 and db.query(KPIAlert).count() == before + 1 + # 幂等：同 (kpi, value, level) 不重复落库 + run_alert_check(db, period=None, dry_run=False, stats={}) + assert db.query(KPIAlert).count() == before + 1 + + +class TestAnalysisApiDimension: + def test_analysis_returns_2d_flag_and_note(self, client, db: Session): + from tests.conftest import create_test_user, get_token_for_user, auth_header + create_test_user(db) + token = get_token_for_user(client) + r = client.post("/api/cma/cashflow-structure/values", headers=auth_header(token), json={ + "period": "2026-08", "operating_net": 14.45, "investing_net": 0.0, + "financing_net": -3.2, "source": "official", "remark": "pytest 裁定1", + }) + assert r.status_code == 200 + cls = r.json()["classification"] + assert cls["classified"] is True and cls["dimension"] == "2D" + assert cls["candidate_type_nos"] == [1, 2] + a = client.get("/api/cma/cashflow-structure/analysis", headers=auth_header(token), + params={"period": "2026-08"}).json() + assert a["classification_dimension"] == "2D" + assert a["investing_note"] == INVESTING_ZERO_NOTE + assert a["nets"]["investing"]["note"] == INVESTING_ZERO_NOTE diff --git a/frontend/src/views/CashFlowStructure.vue b/frontend/src/views/CashFlowStructure.vue index deadfaaa..a516a79c 100644 --- a/frontend/src/views/CashFlowStructure.vue +++ b/frontend/src/views/CashFlowStructure.vue @@ -25,14 +25,24 @@
- 类型{{ classification.type_no }} - {{ classification.health_emoji }} {{ classification.name }} + 二维判定 + 类型{{ classification.type_no }} + 🧭 按 经营(±) × 筹资(±) 二维映射 + {{ classification.health_emoji }} {{ classification.name }}
- 健康度：{{ classification.health_emoji }} {{ classification.health_label }} + 投资活动为 0：{{ investingNote }} + 健康度：{{ classification.health_emoji }} {{ classification.health_label }} 生命周期：{{ classification.lifecycle }} {{ source === 'official' ? '📄 官方现金流量表' : '✍️ 人工录入' }}
+ +
+ + 候选 类型{{ c.type_no }} {{ c.name }}（O{{ c.o_sign }} I{{ c.i_sign }} F{{ c.f_sign }}） + +
@@ -47,6 +57,8 @@
{{ n.sign === '+' ? '进钱 ▲' : n.sign === '-' ? '出钱 ▼' : '为0' }}
+ +
{{ n.note }}
{{ n.hint }}
@@ -67,6 +79,9 @@