cma-management/backend/tests/test_caliber_rulings_20260917.py

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