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 百分比口径
205 lines
11 KiB
Python
205 lines
11 KiB
Python
# -*- coding: utf-8 -*-
|
||
"""财务口径裁定 2026-09-17 落地测试（裁定1 / 2a / 2b）
|
||
|
||
依据：`wecom-finance/reports/CMA_口径裁定_20260917.md`（裁定1 / 2a / 2b）
|
||
+ `wecom-finance/reports/CMA_口径裁定_20260924_比率类单位口径_取代2a两指标.md`
|
||
（`CMA-CALIBER-v2.1/kpi-unit@2026-09-24`：**取代 2a 之 `F_CURRENT_RATIO` / `F_QUICK_RATIO` 两项**
|
||
→ 改「倍」、无量纲直判、不再 ×100；其余比率类仍按 2a 百分比口径）
|
||
· 裁定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（百分比）+ v2.1 取代两项（倍）─────────────
|
||
|
||
class TestRuling2aCaliber:
|
||
def test_current_ratio_multiple_caliber_identity(self):
|
||
"""`F_CURRENT_RATIO` 已按 v2.1 改「倍」口径 → **同量纲直判**（factor=1，不得再 ×100）。
|
||
|
||
反向量纲回归：只改 DB 不改判定层时 1.29×100=129 去比 `>=2.0` 会**恒绿**（比原「恒红」更坏）；
|
||
此处即锁「不被 ×100」这一反向风险。
|
||
"""
|
||
cal = resolve_caliber(1.29, _K("F_CURRENT_RATIO", "倍", ">=2.0", ">=1.0", "<1.0"))
|
||
assert cal["status"] == "ok" and cal["factor"] == 1
|
||
assert cal["value"] == 1.29
|
||
assert cal["note"] is None, "倍口径直判无「口径归一」说明"
|
||
|
||
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_judged_in_multiple_caliber(self, db: Session):
|
||
"""v2.1 两项按「倍」口径**同量纲直判**：1.29 落 [1.0,2.0) → 黄，且消息不带「口径归一」"""
|
||
kpi = create_test_kpi(db, kpi_code="F_CURRENT_RATIO", kpi_name="流动比率", unit="倍",
|
||
threshold_green=">=2.0", threshold_yellow=">=1.0", threshold_red="<1.0")
|
||
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"] == "yellow"
|
||
assert st["planned"][0]["actual"] == 1.29 and st["planned"][0]["factor"] == 1
|
||
assert "口径归一" not in st["planned"][0]["message"]
|
||
assert st["normalized"] == 0
|
||
|
||
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
|