Files
cma-management/backend/tests/test_profitability_indicators.py
Hermes CI Fix 8631ce718f
ci/woodpecker/push/main Pipeline failed
feat(profitability): real ROI (op profit / invested capital), enable EVA, add RI
Kaliber: wecom-finance 2026-10-02 ruling (Boss 11:59 CST "li; but (1) add real ROI").

- kpi_definitions seed scripts/sql/profitability_kpi_defs_20261003.sql:
  F_ROI renamed to F_ROA (same kpi_id=20, all 8 historical values kept), new F_ROI
  (real IMA caliber = operating profit / invested capital), F_EVA activated with unit
  and thresholds aligned to 万元, new F_RI (residual income).
- parameters are NOT hardcoded in formula text: they live in
  kpi_definitions.data_source_config.params (tax_rate / wacc / min_return_rate /
  include_factory_subsidy / include_channel_subsidy / include_sales_discount /
  annualize / capital_basis); single reader = app/utils/profitability_params.py.
- deriver scripts/kpi_profitability_deriver.py: YTD operating profit from
  voucher_details (carry_forward=0 net), invested capital from subject_balances
  level-1 rows only, idempotent upsert into kpi_values (source_type=derived, no
  data_status downgrade, fail-closed when invested capital is 0).
- read-only endpoint GET /reports/value-management reads the derived rows only
  (single source of truth, no recomputation) and exposes the parameter block.
- DupontAnalysis.vue: value management section (ROI / ROA / ROE / EVA / RI).
- caliber_units.py: RATIO_CALIBER_TABLE key F_ROI renamed to F_ROA, new F_ROI row.
- tests: backend/tests/test_profitability_indicators.py (21 cases).
2026-10-03 02:10:47 +08:00

373 lines
18 KiB
Python
Raw Permalink Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
"""盈利能力三指标(真 ROI / EVA / RI)单测 — finance-20261002-profitability-indicators-roi-eva-ri-fullstack
口径依据(财务Bot 2026-10-02 裁定,任总 11:59 CST DM「立;但①要补真ROI」):
营业利润(旧口径) = 5001 − 5401 − 5403 − (5601 + 5602 + 5603)
经营口径营业利润 = 旧口径 + 厂补(530103) − 渠补(571103) [开关可回旧口径]
(571112 商品折扣默认不计入:include_sales_discount=False)
投入资本 = 有息负债(2001 + 2501) + 所有者权益(3xxx **一级科目**净额)
ROI = 营业利润 ÷ 投入资本 × 100 ;NOPAT = 营业利润 × (1 − 所得税率)
EVA = NOPAT − 投入资本 × WACC ;RI = 营业利润 − 投入资本 × 最低报酬率
夹具数值逐项复刻财务复算锚(2026-08,万元):
旧口径营业利润 −267.61 / 厂补 +871.03 / 渠补 −691.43 ⇒ 经营口径 −88.01
投入资本 = 211.50 + 1093.26 = 1304.76
⇒ ROI −6.75% | EVA −170.39 万(NOPAT −66.01) | RI −192.39 万
⇒ include_sales_discount=true(+商品折扣 33.68 万)⇒ −9.33% / −195.65 / −226.07
覆盖:
· app/utils/profitability_params.py —— 参数单一事实源(合并/兜底/兼容形态)
· scripts/kpi_profitability_deriver.py —— 取值 + 计算 + 幂等落库 + fail-closed + 不降级
· app/api/reports.py::/value-management —— 只读端点读派生值(单一事实源)
"""
import datetime
import os
import sys
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
import pytest # noqa: E402
from sqlalchemy import text # noqa: E402
from app.models import KPIDefinition, KPIValue, VoucherDetail # noqa: E402
from app.api.reports import get_value_management # noqa: E402
from app.utils import profitability_params as PP # noqa: E402
from scripts import kpi_profitability_deriver as D # noqa: E402
ENTITY = 1
PERIOD = "2026-08"
D_DATE = datetime.datetime(2026, 8, 31)
# ── 财务复算锚(元)────────────────────────────────────────────────────────
A_SALES_CR = 8_087_955.00 # 5001 贷方累计
A_COST_DR = 8_117_644.00 # 5401
A_TAX_DR = 404_816.00 # 5403
A_5601_DR = 123_144.00
A_5602_DR = 2_062_015.00
A_5603_DR = 56_427.00
A_FACTORY_CR = 8_710_297.00 # 530103 厂补(贷方)
A_CHANNEL_DR = 6_914_297.00 # 571103 渠补(借方)
A_DISCOUNT_DR = 336_793.00 # 571112 商品折扣(借方)
# 2026-08 期末科目余额(元)
A_DEBT_2001 = 2_115_000.00
A_EQ_3001 = 2_000_000.00
A_EQ_3103 = 14_918_101.66
A_EQ_3104_NEG = 5_985_512.16 # 利润分配 借方余额(负权益)
A_EQ_310415_DETAIL = 5_985_512.16 # 3104 的子科目(level=2,须排除防双计)
KPI_CODES = (
("F_ROI", "投入资本回报率(ROI)", "%"),
("F_EVA", "经济增加值(EVA)", "万元"),
("F_RI", "剩余收益(RI)", "万元"),
)
# ── 夹具 ──────────────────────────────────────────────────────────────────
def _create_subject_balances(db):
"""subject_balances 不是 ORM 模型(生产为财务裁定导出的快照表)→ 测试期手工建表。
⚠️ 必须 DROP 后重建:`Base.metadata.drop_all` 不覆盖本表,留表会让行跨用例累积。
"""
db.execute(text("DROP TABLE IF EXISTS subject_balances"))
db.execute(text("""
CREATE TABLE IF NOT EXISTS subject_balances (
id INTEGER PRIMARY KEY AUTOINCREMENT,
entity_id INTEGER NOT NULL,
period VARCHAR(7) NOT NULL,
subject_code VARCHAR(20) NOT NULL,
subject_name VARCHAR(100) NOT NULL DEFAULT '',
level INTEGER NOT NULL,
ob_dr NUMERIC NOT NULL DEFAULT 0, ob_cr NUMERIC NOT NULL DEFAULT 0,
cur_dr NUMERIC NOT NULL DEFAULT 0, cur_cr NUMERIC NOT NULL DEFAULT 0,
ytd_dr NUMERIC NOT NULL DEFAULT 0, ytd_cr NUMERIC NOT NULL DEFAULT 0,
cb_dr NUMERIC NOT NULL DEFAULT 0, cb_cr NUMERIC NOT NULL DEFAULT 0,
entity_scope VARCHAR(24) NOT NULL DEFAULT '',
row_kind VARCHAR(16) NOT NULL DEFAULT '',
source_batch VARCHAR(64),
created_at DATETIME DEFAULT CURRENT_TIMESTAMP
)
"""))
db.commit()
def _sb(db, period, code, level, cb_dr=0.0, cb_cr=0.0, row_kind="subtotal_level1"):
db.execute(text("""
INSERT INTO subject_balances
(entity_id, period, subject_code, subject_name, level, cb_dr, cb_cr, row_kind)
VALUES (:e, :p, :c, :n, :l, :dr, :cr, :rk)
"""), {"e": ENTITY, "p": period, "c": code, "n": code, "l": level,
"dr": cb_dr, "cr": cb_cr, "rk": row_kind})
_seq = [900000]
def _voucher(db, subject_code, debit=0.0, credit=0.0, period=PERIOD, cf=0):
_seq[0] += 1
db.add(VoucherDetail(
entity_id=ENTITY, voucher_no=f"P{_seq[0]}",
period=period, voucher_date=D_DATE, subject_code=subject_code,
subject_name=subject_code, summary="fixture",
debit_amount=debit, credit_amount=credit, carry_forward=cf,
))
def _seed_kpis(db):
for code, name, unit in KPI_CODES:
db.add(KPIDefinition(entity_id=ENTITY, kpi_code=code, kpi_name=name,
dimension="finance", category="profitability",
unit=unit, status="active", target_value=0.0))
db.commit()
def _seed_production_shape(db):
"""按生产 2026-08 形态铺夹具:损益凭证 + 期末科目余额 + 三条 KPI 定义。"""
_create_subject_balances(db)
_seed_kpis(db)
_voucher(db, "5001", credit=A_SALES_CR)
_voucher(db, "5401", debit=A_COST_DR)
_voucher(db, "5403", debit=A_TAX_DR)
_voucher(db, "5601", debit=A_5601_DR)
_voucher(db, "5602", debit=A_5602_DR)
_voucher(db, "5603", debit=A_5603_DR)
_voucher(db, "530103", credit=A_FACTORY_CR)
_voucher(db, "571103", debit=A_CHANNEL_DR)
_voucher(db, "571112", debit=A_DISCOUNT_DR)
_sb(db, PERIOD, "2001", 1, cb_cr=A_DEBT_2001)
_sb(db, PERIOD, "2501", 1)
_sb(db, PERIOD, "3001", 1, cb_cr=A_EQ_3001)
_sb(db, PERIOD, "3103", 1, cb_cr=A_EQ_3103)
_sb(db, PERIOD, "3104", 1, cb_dr=A_EQ_3104_NEG)
_sb(db, PERIOD, "310415", 2, cb_dr=A_EQ_310415_DETAIL, row_kind="detail")
db.commit()
def _values(db, period=PERIOD):
rows = db.query(KPIValue, KPIDefinition).join(
KPIDefinition, KPIDefinition.id == KPIValue.kpi_id).filter(
KPIValue.period == period).all()
return {k.kpi_code: v for v, k in rows}
# ── 参数单一事实源 ─────────────────────────────────────────────────────────
class TestParams:
def test_defaults_when_no_definitions(self, db):
p = PP.load_profitability_params(db, ENTITY)
assert p["tax_rate"] == 0.25 and p["wacc"] == 0.08
assert p["min_return_rate"] == 0.08
assert p["include_factory_subsidy"] is True
assert p["include_channel_subsidy"] is True
assert p["include_sales_discount"] is False
assert p["annualize"] is False and p["capital_basis"] == "ending"
def test_parse_variants(self):
assert PP.parse_params(None) == {}
assert PP.parse_params("") == {}
assert PP.parse_params('{"params": {"wacc": 0.1}}') == {"wacc": 0.1}
assert PP.parse_params({"wacc": 0.1}) == {"wacc": 0.1}
assert PP.parse_params("not-json") == {}
assert PP.parse_params('{"params": null}') == {}
assert PP.parse_params({"params": {"x": None}}) == {}
def test_merge_three_rows_over_defaults(self, db):
db.add(KPIDefinition(entity_id=ENTITY, kpi_code="F_ROI", kpi_name="ROI", unit="%",
data_source_config={"params": {"include_sales_discount": True}}))
db.add(KPIDefinition(entity_id=ENTITY, kpi_code="F_EVA", kpi_name="EVA", unit="万元",
data_source_config={"params": {"wacc": 0.12, "tax_rate": 0.15}}))
db.add(KPIDefinition(entity_id=ENTITY, kpi_code="F_RI", kpi_name="RI", unit="万元",
data_source_config={"params": {"min_return_rate": 0.11}}))
db.commit()
p = PP.load_profitability_params(db, ENTITY)
assert p["wacc"] == 0.12 and p["tax_rate"] == 0.15
assert p["min_return_rate"] == 0.11
assert p["include_sales_discount"] is True
# 未覆盖项仍取默认
assert p["include_factory_subsidy"] is True and p["capital_basis"] == "ending"
# ── 纯计算(口径锚) ───────────────────────────────────────────────────────
class TestCompute:
def test_finance_anchor_values(self):
v = D.compute_indicators(-88.0091, 1304.75895, 8, dict(PP.DEFAULTS))
assert v["F_ROI"] == pytest.approx(-6.74, abs=0.02)
assert v["nopat_w"] == pytest.approx(-66.0068, abs=0.01)
assert v["F_EVA"] == pytest.approx(-170.4, abs=0.1)
assert v["F_RI"] == pytest.approx(-192.4, abs=0.1)
def test_switch_sales_discount_moves_all_three(self):
base = D.compute_indicators(-88.0091, 1304.75895, 8, dict(PP.DEFAULTS))
p2 = dict(PP.DEFAULTS, include_sales_discount=True)
with_disc = D.compute_indicators(-121.6884, 1304.75895, 8, p2)
assert (base["F_ROI"], base["F_EVA"], base["F_RI"]) != \
(with_disc["F_ROI"], with_disc["F_EVA"], with_disc["F_RI"])
assert with_disc["F_ROI"] == pytest.approx(-9.33, abs=0.02)
assert with_disc["F_EVA"] == pytest.approx(-195.6, abs=0.2)
assert with_disc["F_RI"] == pytest.approx(-225.8, abs=0.5)
def test_annualize_changes_roi(self):
p = dict(PP.DEFAULTS, annualize=True)
v = D.compute_indicators(-88.0091, 1304.75895, 8, p)
assert v["scale"] == pytest.approx(1.5)
assert v["F_ROI"] == pytest.approx(-10.12, abs=0.05)
def test_fail_closed_when_capital_zero(self):
v = D.compute_indicators(-88.0, 0.0, 8, dict(PP.DEFAULTS))
assert v["F_ROI"] is None and v["F_EVA"] is None and v["F_RI"] is None
# ── 派生器取值 / 落库 ──────────────────────────────────────────────────────
class TestDeriver:
def test_production_shape_matches_finance_anchor(self, db):
_seed_production_shape(db)
r = D.run_derive(db, ENTITY, PERIOD, write_kpi=True)
p, c = r["profit"], r["capital"]
assert p["sales_w"] == pytest.approx(808.7955, abs=1e-3)
assert p["op_profit_legacy_w"] == pytest.approx(-267.61, abs=0.01)
assert p["factory_subsidy_w"] == pytest.approx(871.03, abs=0.01)
assert p["channel_subsidy_w"] == pytest.approx(691.43, abs=0.01)
assert p["op_profit_w"] == pytest.approx(-88.01, abs=0.01)
# 一级科目口径:3104 与子科目 310415 不得双计
assert c["interest_bearing_debt_w"] == pytest.approx(211.50, abs=0.01)
assert c["equity_w"] == pytest.approx(1093.26, abs=0.01)
assert c["invest_capital_w"] == pytest.approx(1304.76, abs=0.01)
assert r["values"]["F_ROI"] == pytest.approx(-6.74, abs=0.02)
assert r["values"]["F_EVA"] == pytest.approx(-170.4, abs=0.1)
assert r["values"]["F_RI"] == pytest.approx(-192.4, abs=0.1)
assert r["written"] == {"F_ROI": "insert", "F_EVA": "insert", "F_RI": "insert"}
def test_rows_written_with_derived_source(self, db):
_seed_production_shape(db)
D.run_derive(db, ENTITY, PERIOD, write_kpi=True)
vals = _values(db)
assert set(vals) == {"F_ROI", "F_EVA", "F_RI"}
for code, row in vals.items():
assert row.source_type == "derived"
assert row.entity_id == ENTITY and row.period == PERIOD
assert row.actual_value is not None and row.source_batch.startswith("profit-")
assert "投入资本" in (row.remark or "")
def test_idempotent_rerun_no_duplicate(self, db):
_seed_production_shape(db)
D.run_derive(db, ENTITY, PERIOD, write_kpi=True)
r2 = D.run_derive(db, ENTITY, PERIOD, write_kpi=True)
assert r2["written"] == {"F_ROI": "upsert", "F_EVA": "upsert", "F_RI": "upsert"}
assert r2["changed"] == [] # 值未变 ⇒ 无值变留痕
n = db.query(KPIValue).filter(KPIValue.source_type == "derived").count()
assert n == 3
def test_does_not_downgrade_verified_status(self, db):
_seed_production_shape(db)
kpi = db.query(KPIDefinition).filter(KPIDefinition.kpi_code == "F_ROI").first()
db.add(KPIValue(entity_id=ENTITY, kpi_id=kpi.id, period=PERIOD, actual_value=-99.0,
source_type="derived", source_batch="old", data_status="verified"))
db.commit()
r = D.run_derive(db, ENTITY, PERIOD, write_kpi=True)
row = _values(db)["F_ROI"]
assert row.data_status == "verified" # 不降级(ASK-20260924-10)
assert row.actual_value == pytest.approx(-6.74, abs=0.02)
assert r["changed"] and r["changed"][0]["code"] == "F_ROI"
def test_switch_true_changes_values(self, db):
_seed_production_shape(db)
D.run_derive(db, ENTITY, PERIOD, write_kpi=True)
base = {c: v.actual_value for c, v in _values(db).items()}
kpi = db.query(KPIDefinition).filter(KPIDefinition.kpi_code == "F_ROI").first()
kpi.data_source_config = {"params": {"include_sales_discount": True}}
db.commit()
D.run_derive(db, ENTITY, PERIOD, write_kpi=True)
now = {c: v.actual_value for c, v in _values(db).items()}
assert now["F_ROI"] == pytest.approx(-9.33, abs=0.02)
assert now["F_EVA"] == pytest.approx(-195.6, abs=0.2)
assert now["F_RI"] == pytest.approx(-225.8, abs=0.5)
assert now != base
def test_wacc_change_moves_eva_only(self, db):
_seed_production_shape(db)
D.run_derive(db, ENTITY, PERIOD, write_kpi=True)
base = {c: v.actual_value for c, v in _values(db).items()}
eva = db.query(KPIDefinition).filter(KPIDefinition.kpi_code == "F_EVA").first()
eva.data_source_config = {"params": {"tax_rate": 0.25, "wacc": 0.10}}
db.commit()
D.run_derive(db, ENTITY, PERIOD, write_kpi=True)
now = {c: v.actual_value for c, v in _values(db).items()}
assert now["F_ROI"] == base["F_ROI"] and now["F_RI"] == base["F_RI"]
# WACC 8%→10% ⇒ EVA 少 1304.75895 × 2% = 26.10 万
assert now["F_EVA"] == pytest.approx(base["F_EVA"] - 26.10, abs=0.02)
def test_fail_closed_without_capital_no_write(self, db):
_seed_production_shape(db)
db.execute(text("DELETE FROM subject_balances"))
db.commit()
r = D.run_derive(db, ENTITY, PERIOD, write_kpi=True)
assert r["capital"]["invest_capital_w"] == 0.0
assert r["values"] == {"F_ROI": None, "F_EVA": None, "F_RI": None}
assert db.query(KPIValue).filter(KPIValue.source_type == "derived").count() == 0
def test_missing_definition_reported_not_crash(self, db):
_create_subject_balances(db)
_seed_kpis(db)
_voucher(db, "5001", credit=A_SALES_CR)
_sb(db, PERIOD, "3001", 1, cb_cr=A_EQ_3001)
ri = db.query(KPIDefinition).filter(KPIDefinition.kpi_code == "F_RI").first()
db.delete(ri)
db.commit()
r = D.run_derive(db, ENTITY, PERIOD, write_kpi=True)
assert r["written"]["F_RI"] == "no-definition"
assert r["written"]["F_ROI"] == "insert"
def test_pick_period_uses_latest_subject_balances(self, db):
_create_subject_balances(db)
_sb(db, "2026-07", "3001", 1, cb_cr=A_EQ_3001)
_sb(db, "2026-08", "3001", 1, cb_cr=A_EQ_3001)
db.commit()
assert D._pick_period(db, ENTITY) == "2026-08"
def test_average_basis_falls_back_when_prev_missing(self, db):
_seed_production_shape(db)
c = D.invest_capital(db, ENTITY, PERIOD, "average")
assert c["basis"] == "ending(fallback)"
assert c["invest_capital_w"] == pytest.approx(1304.76, abs=0.01)
def test_average_basis_uses_prev_period(self, db):
_seed_production_shape(db)
_sb(db, "2026-07", "2001", 1, cb_cr=A_DEBT_2001)
_sb(db, "2026-07", "3001", 1, cb_cr=A_EQ_3001)
db.commit()
c = D.invest_capital(db, ENTITY, PERIOD, "average")
assert c["basis"] == "average"
# 上一期 (2026-07) = 有息负债 211.50万 + 权益 200.00万 = 411.50万
assert c["prev_total_w"] == pytest.approx(411.50, abs=0.01)
# (411.50 + 1304.75895) / 2
assert c["invest_capital_w"] == pytest.approx((411.5 + 1304.75895) / 2, abs=0.01)
# ── 只读端点(读派生值单一事实源) ─────────────────────────────────────────
class TestValueManagementEndpoint:
def test_reads_derived_values(self, db):
_seed_production_shape(db)
D.run_derive(db, ENTITY, PERIOD, write_kpi=True)
r = get_value_management(period=PERIOD, db=db, entity_id=ENTITY)
assert r["readonly"] is True and r["data_source"] == "derived"
assert r["period"] == PERIOD
assert r["roi"]["value"] == pytest.approx(-6.74, abs=0.02)
assert r["eva"]["value"] == pytest.approx(-170.4, abs=0.1)
assert r["ri"]["value"] == pytest.approx(-192.4, abs=0.1)
assert r["roa"] is None # 未 seed F_ROA
assert r["params"]["wacc"] == 0.08
assert r["params"]["include_sales_discount"] is False
def test_default_period_is_latest(self, db):
_seed_production_shape(db)
D.run_derive(db, ENTITY, PERIOD, write_kpi=True)
r = get_value_management(period=None, db=db, entity_id=ENTITY)
assert r["period"] == PERIOD
def test_empty_when_no_rows(self, db):
_seed_kpis(db)
r = get_value_management(period=None, db=db, entity_id=ENTITY)
assert r["period"] is None and r["roi"] is None and r["params"]["tax_rate"] == 0.25