# -*- 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