diff --git a/backend/app/api/reports.py b/backend/app/api/reports.py index eebd254c..1dbb3dfd 100644 --- a/backend/app/api/reports.py +++ b/backend/app/api/reports.py @@ -3,7 +3,7 @@ CMA管理报表中心 — 管理会计OS 非传统财务报表，聚焦管理决策分析 报表: - 1. 管理利润表 — 收入→变动成本→边际贡献→固定成本→息税前利润 + 1. 管理利润表 — 收入→变动成本→边际贡献→固定成本→营业利润（真实科目取数） 2. 预算执行报告 — 各KPI预算vs实际vs差异率 3. KPI趋势报告 — 选定KPI的历史趋势 4. 四维度绩效评分卡 — BSC健康度雷达图 @@ -28,6 +28,8 @@ from app.utils.dupont_basis import ( resolve_window, stock_balance, sum_flows, ) from app.utils.voucher_guard import apply_voucher_guard +from app.utils.profit_lines import (BOTTOM_LABEL, UNIT, build_items, compute_lines, + prev_period_of, reconcile) from app.services import report_reclass from app.deps import get_entity_id, assert_entity_slug_scope, resolve_session_slug import logging @@ -49,116 +51,41 @@ def get_profit_summary( db: Session = Depends(get_db), entity_id: int = Depends(get_entity_id), ): - """管理利润表 — 收入→变动成本→边际贡献→固定成本→息税前利润""" + """管理利润表 — 收入→变动成本→边际贡献→固定成本→营业利润（**真实科目取数**，按 entity 隔离） + + 2026-10-14 重建（原实现 3 处硬伤已消除）： + ① 原 L74 `contribution_margin = gross_profit`（边际贡献=毛利）→ 改真账：收入 − 变动成本； + ② 原 L76 变动成本 = 营收 × 硬编码比例（五成）→ 科目口径现算，**本端点零比例常量**； + ③ 原末行标签「＝ 息税前利润」却装净利率算出的 net_profit → 标签=算法=「营业利润」 + （＝ 收入 − 变动成本 − 固定成本；未扣所得税/营业外 ⇒ 不称净利润，已扣 6603 财务费用 ⇒ 不称 EBIT）。 + + 口径唯一源：`app/utils/bep_caliber.py::ENTITY_CALIBERS`（变动/固定科目构成一律从该处派生， + 本报表层不定义任何科目集）；取数实现见 `app/utils/profit_lines.py`（voucher_details 真实发生额）。 + 勾稽自检：响应 `reconciliation`（收入 − 变动成本 − 固定成本 = 末行；不平显式标出，禁静默不平）。 + """ if period is None: period = datetime.now().strftime("%Y-%m") + prev_period = prev_period_of(period) - # 从KPI数据中获取各利润要素 - def get_val(code: str): - kpi = db.query(KPIDefinition).filter( - KPIDefinition.kpi_code == code, KPIDefinition.entity_id == entity_id).first() - if not kpi: - return None - v = db.query(KPIValue).filter( - KPIValue.kpi_id == kpi.id, KPIValue.period == period - ).order_by(KPIValue.id.desc()).first() - return v.actual_value if v else None - - revenue = get_val("F_REVENUE") - gross_profit_rate = get_val("F_PROFIT_RATE") - net_profit_rate = get_val("F_NET_PROFIT_RATE") - cost_ratio = get_val("F_COST_RATIO") - - # 计算利润要素 - # 营收已知，用毛利率算毛利，用成本率算成本 - gross_profit = round(revenue * (gross_profit_rate / 100), 2) if revenue and gross_profit_rate else None - total_cost = round(revenue * (cost_ratio / 100), 2) if revenue and cost_ratio else None - net_profit = round(revenue * (net_profit_rate / 100), 2) if revenue and net_profit_rate else None - - # 边际贡献 ≈ 毛利（简化模型） - contribution_margin = gross_profit - # 固定成本 ≈ 总成本 - 变动成本（假设变动成本=营收*50%） - variable_cost = round(revenue * 0.50, 2) if revenue else None - fixed_cost = round(total_cost - variable_cost, 2) if total_cost and variable_cost else None - - # 找上期做环比 - prev_year, prev_month = period.split("-") - py, pm = int(prev_year), int(prev_month) - pm -= 1 - if pm <= 0: - pm += 12 - py -= 1 - prev_period = f"{py}-{pm:02d}" - - def get_prev_val(code: str): - kpi = db.query(KPIDefinition).filter( - KPIDefinition.kpi_code == code, KPIDefinition.entity_id == entity_id).first() - if not kpi: return None - v = db.query(KPIValue).filter( - KPIValue.kpi_id == kpi.id, KPIValue.period == prev_period - ).order_by(KPIValue.id.desc()).first() - return v.actual_value if v else None - - prev_revenue = get_prev_val("F_REVENUE") - prev_gross_profit_rate = get_prev_val("F_PROFIT_RATE") - prev_net_profit_rate = get_prev_val("F_NET_PROFIT_RATE") - prev_cost_ratio = get_prev_val("F_COST_RATIO") - prev_gross_profit = round(prev_revenue * (prev_gross_profit_rate / 100), 2) if prev_revenue and prev_gross_profit_rate else None - prev_total_cost = round(prev_revenue * (prev_cost_ratio / 100), 2) if prev_revenue and prev_cost_ratio else None - prev_net_profit = round(prev_revenue * (prev_net_profit_rate / 100), 2) if prev_revenue and prev_net_profit_rate else None - prev_contribution_margin = prev_gross_profit - prev_variable_cost = round(prev_revenue * 0.50, 2) if prev_revenue else None - prev_fixed_cost = round(prev_total_cost - prev_variable_cost, 2) if prev_total_cost and prev_variable_cost else None - - def calc_chg(cur, prev): - if cur is not None and prev is not None and prev != 0: - return round((cur - prev) / prev * 100, 2) - return None - - items = [ - { - "name": "营业收入", - "value": revenue, - "prev_value": prev_revenue, - "change_rate": calc_chg(revenue, prev_revenue), - "ratio": 100.0, - }, - { - "name": "减：变动成本", - "value": variable_cost, - "prev_value": prev_variable_cost, - "change_rate": calc_chg(variable_cost, prev_variable_cost), - "ratio": round(variable_cost / revenue * 100, 2) if variable_cost and revenue else None, - }, - { - "name": "＝ 边际贡献", - "value": contribution_margin, - "prev_value": prev_contribution_margin, - "change_rate": calc_chg(contribution_margin, prev_contribution_margin), - "ratio": round(contribution_margin / revenue * 100, 2) if contribution_margin and revenue else None, - "is_subtotal": True, - }, - { - "name": "减：固定成本", - "value": fixed_cost, - "prev_value": prev_fixed_cost, - "change_rate": calc_chg(fixed_cost, prev_fixed_cost), - "ratio": round(fixed_cost / revenue * 100, 2) if fixed_cost and revenue else None, - }, - { - "name": "＝ 息税前利润", - "value": net_profit, - "prev_value": prev_net_profit, - "change_rate": calc_chg(net_profit, prev_net_profit), - "ratio": round(net_profit / revenue * 100, 2) if net_profit and revenue else None, - "is_total": True, - }, - ] + cur = compute_lines(db, entity_id, period, period) + prev = compute_lines(db, entity_id, prev_period, prev_period) + values = cur.get("values") or {} return { "period": period, "prev_period": prev_period, - "items": items, + "entity_id": entity_id, + "unit": UNIT, + "bottom_label": BOTTOM_LABEL, + "source": "voucher_details 真实凭证明细发生额（按 entity 隔离；剔结转腿 + 作废/撤单行）", + "available": cur["available"], + "missing": cur["missing"], + "prev_available": prev["available"], + "prev_missing": prev["missing"], + "caliber": cur.get("caliber"), + "items": build_items(cur, prev), + "reconciliation": reconcile(values.get("revenue"), values.get("variable_cost"), + values.get("fixed_cost"), values.get("result")), } diff --git a/backend/app/utils/profit_lines.py b/backend/app/utils/profit_lines.py new file mode 100644 index 00000000..ff43be00 --- /dev/null +++ b/backend/app/utils/profit_lines.py @@ -0,0 +1,371 @@ +# -*- coding: utf-8 -*- +"""管理利润表（贡献式）取数层 —— 真实科目发生额，按账套(entity)隔离。 + +**口径唯一源**：`app/utils/bep_caliber.py::ENTITY_CALIBERS` + · 收入 ：`caliber['revenue_field']` + `caliber['revenue_filter']`（e1=5001 贷方 / e2=6001 贷方） + · 变动成本 ：`caliber['varcost_terms']`（带号科目项；e1 = 5401借 + 5403借 + 571103借 − 530103贷） + · 固定成本 ：`caliber['fixed_terms']`（带号科目项，**优先**）否则 `caliber['fixed_filters']` + （前缀元组 → 派生为 `('+', '<前缀>%', 'net')` 借−贷净额；口径升级为 `fixed_terms` 后本层零改动自动跟随） + · 期间行域 ：`carry_forward == 0` + `caliber['exclude_closing']`（结转腿未标账套须显式剔 `summary LIKE '%结转%'`） + · 作废行域 ：`apply_voucher_guard`（`voucher_void_marks`，全库唯一实现） + +铁律（2026-10-14 管理利润表重建单） + 1. 本模块**不定义任何科目集、不含任何比例常量**；科目构成一律从 ENTITY_CALIBERS 派生。 + 口径要改 → 只改 `bep_caliber.py` 一处（本层与报表层零改动自动跟随），**禁第二套口径**。 + 2. **禁** KPI 比率 × 收入倒推（旧实现 `F_PROFIT_RATE` / `F_NET_PROFIT_RATE` / `F_COST_RATIO` 路线已废）； + **禁**硬编码比例（旧实现：变动成本 = 营收 × 写死的五成比例）。 + 3. 舍入一律「**汇总后舍入**」：先按元聚合相减，末步 ÷10000 取 2 位；勿逐分量舍入后相加。 + 4. 查不到的数据**如实标缺**（`None` + `missing` 列表），不倒推、不回落政策常量。 + 5. 末行标签必须与算法一致：本层算的是「收入 − 变动成本 − 固定成本」⇒ 标签 `营业利润` + （常量 `BOTTOM_LABEL`）。**不称净利润**（未扣所得税/营业外收支/投资收益）； + **不称息税前利润**（已扣 6603 财务费用，EBIT 应在其之上）。 +""" +from __future__ import annotations + +import hashlib +import json +from typing import Dict, List, Optional, Tuple + +from sqlalchemy import func, or_ + +from app.models import VoucherDetail +from app.utils.bep_caliber import CLOSING_SUMMARY_LIKE, caliber_or_none +from app.utils.voucher_guard import apply_voucher_guard + +# ── 展示口径 ──────────────────────────────────────────────────────────────── +UNIT = "万元" # 库内为元，展示 ÷10,000（与 bep_caliber 头注一致） +YUAN_PER_WAN = 10000.0 +ROUND_DP = 2 +# 勾稽容差（万元）：三分量各 2 位舍入，「汇总后舍入」口径下最大漂移 < 0.015 ⇒ 0.02 +RECON_TOL_WAN = 0.02 + +# 末行标签（与算法一致：收入 − 变动成本 − 固定成本） +BOTTOM_LABEL = "营业利润" +BOTTOM_LABEL_NOTE = ("＝ 营业收入 − 变动成本 − 固定成本；不含营业外收支、投资收益、所得税 " + "⇒ 不标「净利润」；已扣 6603 财务费用 ⇒ 不标「息税前利润」") +RECON_FORMULA = "营业收入 − 变动成本 − 固定成本 = 末行" + +FIELD_MAP = { + "debit": lambda: VoucherDetail.debit_amount, + "credit": lambda: VoucherDetail.credit_amount, + "net": lambda: VoucherDetail.debit_amount - VoucherDetail.credit_amount, # 借−贷 + # 列名别名（与 dashboard._bep_sum_field / kpi_bep_deriver._sum_field 的调用风格互通，防两套拼写） + "debit_amount": lambda: VoucherDetail.debit_amount, + "credit_amount": lambda: VoucherDetail.credit_amount, +} +_FIELD_FROM_COLUMN = {"debit_amount": "debit", "credit_amount": "credit", "net": "net"} + +# 口径指纹涉及的键（变更即变；供报表/回执标注取证） +_DIGEST_KEYS = ("name", "varcost_terms", "fixed_terms", "fixed_filters", "revenue_filter", + "revenue_field", "exclude_closing", "closing_subject") + + +# ── 通用聚合原语（机制层；与 dashboard._bep_sum_field / kpi_bep_deriver._sum_field 同式同参）── + +def subject_clause(kind: str, code: str): + """科目过滤谓词：('eq', code) 精确 / ('like', prefix) 前缀。""" + if kind == "eq": + return VoucherDetail.subject_code == code + if kind == "like": + return VoucherDetail.subject_code.like(str(code) + "%") + raise ValueError(f"未知科目过滤类型: {kind}") + + +def sum_by_subject(db, entity_id: int, period_from: str, period_to: str, field: str, + subject_filter: Optional[Tuple[str, str]] = None, *, + subject_filters: Optional[List[Tuple[str, str]]] = None, + exclude_closing: bool = False) -> Dict[str, float]: + """按科目分组的金额聚合（**元**）→ {subject_code: 金额}。 + + · field ∈ {debit, credit, net(=借−贷)}；`net` 只在 `carry_forward==0` 行域成立 + · subject_filter = ('eq', code) / ('like', prefix)；subject_filters = [(kind, code), ...] 多科目 OR + · 作废/撤单行经 apply_voucher_guard 剔除（标记表缺失时原样放行并告警一次） + · 返回按科目明细 ⇒ 报表每行可逐行对到真实科目（取数留痕） + """ + expr = FIELD_MAP[field]() + q = db.query(VoucherDetail.subject_code, func.sum(expr)).filter( + VoucherDetail.entity_id == entity_id, + VoucherDetail.carry_forward == 0, + VoucherDetail.period >= period_from, + VoucherDetail.period <= period_to, + ) + if subject_filters: + q = q.filter(or_(*[subject_clause(k, c) for k, c in subject_filters])) + elif subject_filter: + q = q.filter(subject_clause(*subject_filter)) + if exclude_closing: + q = q.filter(or_(VoucherDetail.summary.is_(None), + ~VoucherDetail.summary.like(CLOSING_SUMMARY_LIKE))) + q = q.group_by(VoucherDetail.subject_code).order_by(VoucherDetail.subject_code) + q = apply_voucher_guard(db, q, VoucherDetail) + return {str(code): float(amount or 0.0) for code, amount in q.all()} + + +def sum_field(db, entity_id: int, period_from: str, period_to: str, field: str, + subject_filter: Optional[Tuple[str, str]] = None, *, + subject_filters: Optional[List[Tuple[str, str]]] = None, + exclude_closing: bool = False) -> float: + """单科目/多科目金额合计（元）—— `sum_by_subject` 的标量包装（供口径对拍）。""" + return float(sum(sum_by_subject(db, entity_id, period_from, period_to, field, + subject_filter, subject_filters=subject_filters, + exclude_closing=exclude_closing).values())) + + +def sum_terms(db, entity_id: int, period_from: str, period_to: str, terms, *, + exclude_closing: bool = False) -> dict: + """「带号科目项」合计（元）—— terms = [(sign, code, field), ...]。 + + · sign='+' 加该科目项发生额；'-' 减之 + · code 以 '%' 结尾 ⇒ 前缀匹配（like），否则精确匹配（eq） + · 返回 {'total_yuan': float, 'parts': [{'sign','code','field','match','total_yuan','by_subject'}]} + `parts` 即逐行取数留痕（可逐项对到真实科目与金额） + """ + total = 0.0 + parts: List[dict] = [] + for sign, code, field in terms: + if sign not in ("+", "-"): + raise ValueError(f"未知符号: {sign}（只支持 +/-）") + raw = str(code) + kind = "like" if raw.endswith("%") else "eq" + code_arg = raw[:-1] if kind == "like" else raw + by = sum_by_subject(db, entity_id, period_from, period_to, field, + (kind, code_arg), exclude_closing=exclude_closing) + sub = float(sum(by.values())) + total += sub if sign == "+" else -sub + parts.append({ + "sign": sign, "code": raw, "field": field, "match": kind, + "total_yuan": round(sub, ROUND_DP), + "by_subject": {k: round(v, 2) for k, v in by.items()}, + }) + return {"total_yuan": total, "parts": parts} + + +def has_voucher_rows(db, entity_id: int, period_from: str, period_to: str, + exclude_closing: bool = False) -> bool: + """该账套/期间行域内是否有明细行（无 ⇒ 报表如实标缺，而不是当作 0）。""" + q = db.query(func.count(VoucherDetail.id)).filter( + VoucherDetail.entity_id == entity_id, + VoucherDetail.carry_forward == 0, + VoucherDetail.period >= period_from, + VoucherDetail.period <= period_to, + ) + if exclude_closing: + q = q.filter(or_(VoucherDetail.summary.is_(None), + ~VoucherDetail.summary.like(CLOSING_SUMMARY_LIKE))) + q = apply_voucher_guard(db, q, VoucherDetail) + return bool(q.scalar() or 0) + + +# ── 口径派生（全部来自 ENTITY_CALIBERS，本层零科目集） ─────────────────────── + +def revenue_terms(cal: dict) -> list: + """收入项（带号科目项）：由 caliber['revenue_field'/'revenue_filter'] 派生。""" + kind, code = cal["revenue_filter"] + field = _FIELD_FROM_COLUMN.get(cal.get("revenue_field") or "credit_amount", "credit") + return [("+", (str(code) + "%") if kind == "like" else str(code), field)] + + +def fixed_cost_terms(cal: dict) -> Tuple[list, str]: + """固定成本科目项：优先 `fixed_terms`（带号项），否则由 `fixed_filters` 派生（前缀 + 借−贷净额）。 + + 返回 (terms, mode)；mode 供报表标注当前口径形态（`fixed_terms` / `fixed_filters(net)`）。 + """ + terms = cal.get("fixed_terms") + if terms: + return [tuple(t) for t in terms], "fixed_terms" + derived = [("+", str(code) + "%", "net") for _kind, code in (cal.get("fixed_filters") or [])] + return derived, "fixed_filters(net)" + + +def caliber_digest(cal: dict) -> str: + """口径指纹（md5 12 位）—— 口径表变更即变，供报表/回执标注取证。""" + payload = json.dumps({k: cal.get(k) for k in _DIGEST_KEYS}, + ensure_ascii=False, sort_keys=True, default=str) + return hashlib.md5(payload.encode("utf-8")).hexdigest()[:12] + + +def prev_period_of(period: str) -> str: + """上期（月）—— '2026-01' → '2025-12'。""" + year, month = str(period).split("-")[:2] + y, m = int(year), int(month) + m -= 1 + if m <= 0: + m += 12 + y -= 1 + return f"{y}-{m:02d}" + + +# ── 取数主入口 ────────────────────────────────────────────────────────────── + +def compute_lines(db, entity_id: int, period_from: str, period_to: str) -> dict: + """按 entity 口径从**真实科目发生额**算管理利润表各行（元 + 万元）。 + + 返回: + available 口径已注册且有明细行（否则 False + missing 说明，禁倒推） + caliber 口径快照（含 digest / 各科目项 ⇒ 报表可逐行对到科目） + raw_yuan 元级原值（revenue/variable_cost/contribution/fixed_cost/result） + values 万元（2 位，**汇总后舍入**：先元级相减，末步落位） + rows 各行的带号科目项明细（parts/by_subject，取数留痕） + """ + out = { + "entity_id": entity_id, "period_from": period_from, "period_to": period_to, + "available": False, "missing": [], + "raw_yuan": {}, "values": {k: None for k in + ("revenue", "variable_cost", "contribution", "fixed_cost", "result")}, + "rows": {}, "caliber": None, + } + cal = caliber_or_none(entity_id) + if cal is None: + out["missing"].append( + f"entity={entity_id} 未在 ENTITY_CALIBERS 注册口径（禁编造/禁跨账套套用：" + f"固定成本与收入基数按账套而异，见 app/utils/bep_caliber.py）") + return out + + exclude_closing = bool(cal.get("exclude_closing")) + fix_terms, fix_mode = fixed_cost_terms(cal) + rev_terms = revenue_terms(cal) + out["caliber"] = { + "entity_id": entity_id, + "name": cal.get("name"), + "margin_label": cal.get("margin_label"), + "margin_desc": cal.get("margin_desc"), + "revenue_terms": [list(t) for t in rev_terms], + "varcost_terms": [list(t) for t in (cal.get("varcost_terms") or [])], + "fixed_cost_terms": [list(t) for t in fix_terms], + "fixed_mode": fix_mode, + "exclude_closing": exclude_closing, + "closing_subject": cal.get("closing_subject"), + "digest": caliber_digest(cal), + "unit": UNIT, + } + + if not has_voucher_rows(db, entity_id, period_from, period_to, exclude_closing): + out["missing"].append( + f"entity={entity_id} 期间 {period_from}..{period_to} 无凭证明细行（如实标缺，不倒推）") + return out + + rev = sum_terms(db, entity_id, period_from, period_to, rev_terms, exclude_closing=exclude_closing) + var = sum_terms(db, entity_id, period_from, period_to, cal.get("varcost_terms") or [], + exclude_closing=exclude_closing) + fix = sum_terms(db, entity_id, period_from, period_to, fix_terms, exclude_closing=exclude_closing) + + rev_y = rev["total_yuan"] + var_y = var["total_yuan"] + fix_y = fix["total_yuan"] + con_y = rev_y - var_y # 边际贡献 + res_y = con_y - fix_y # 末行（营业利润） + + out["rows"] = {"revenue": rev, "variable_cost": var, "fixed_cost": fix} + out["raw_yuan"] = {"revenue": round(rev_y, 2), "variable_cost": round(var_y, 2), + "contribution": round(con_y, 2), "fixed_cost": round(fix_y, 2), + "result": round(res_y, 2)} + # 汇总后舍入：元级相减后一次性落万元，不逐分量舍入相加 + out["values"] = {"revenue": round(rev_y / YUAN_PER_WAN, ROUND_DP), + "variable_cost": round(var_y / YUAN_PER_WAN, ROUND_DP), + "contribution": round(con_y / YUAN_PER_WAN, ROUND_DP), + "fixed_cost": round(fix_y / YUAN_PER_WAN, ROUND_DP), + "result": round(res_y / YUAN_PER_WAN, ROUND_DP)} + out["available"] = True + return out + + +def reconcile(revenue: Optional[float], variable_cost: Optional[float], + fixed_cost: Optional[float], result: Optional[float], + tol: float = RECON_TOL_WAN) -> dict: + """勾稽自检（纯函数）：**收入 − 变动成本 − 固定成本 = 末行**。 + + 不平 ⇒ `balanced=False` + `diff`（报表层必须显式标出，**不许静默不平**）； + 组件缺数 ⇒ `balanced=False` + `components_present=False` + `missing`（同样不静默）。 + """ + names = (("revenue", revenue), ("variable_cost", variable_cost), + ("fixed_cost", fixed_cost), ("result", result)) + missing = [n for n, v in names if v is None] + if missing: + return {"balanced": False, "diff": None, "lhs": None, "rhs": result, + "tolerance": tol, "components_present": False, "missing": missing, + "formula": RECON_FORMULA, + "note": "组件缺数 ⇒ 无法勾稽（显式标缺，不静默平）"} + # 上行已排除 None（missing 为空）⇒ 此处断言收敛类型，供静态检查/防回归 + assert revenue is not None and variable_cost is not None \ + and fixed_cost is not None and result is not None + lhs = round(revenue - variable_cost - fixed_cost, ROUND_DP) + diff = round(lhs - result, ROUND_DP) + balanced = abs(diff) <= tol + return {"balanced": balanced, "diff": diff, "lhs": lhs, "rhs": result, + "tolerance": tol, "components_present": True, "missing": [], + "formula": RECON_FORMULA, + "note": ("收入 − 变动成本 − 固定成本 = 末行 ✓" if balanced else + f"⚠️ 勾稽不平：收入−变动−固定={lhs}，末行={result}，差 {diff}（万元）")} + + +def _calc_chg(cur: Optional[float], prev: Optional[float]) -> Optional[float]: + if cur is not None and prev is not None and prev != 0: + return round((cur - prev) / prev * 100, 2) + return None + + +def build_items(cur: dict, prev: dict, bottom_label: str = BOTTOM_LABEL) -> list: + """按管理利润表结构组 5 行（收入/变动成本/边际贡献/固定成本/末行），含环比与占比。 + + 每行附 `terms`（带号科目项）+ `raw_yuan`（元级）⇒ 页面/回执可**逐行对到真实科目**。 + 口径未注册或无明细行 ⇒ 各行 value 为 None（如实标缺）。 + """ + cv = cur.get("values") or {} + pv = prev.get("values") or {} + cre = cur.get("rows") or {} + pre = prev.get("rows") or {} + + def terms_of(key: str, src: dict): + row = (cre if src is cur else pre).get(key) + return row.get("parts") if row else None + + def raw(key: str) -> Optional[float]: + return (cur.get("raw_yuan") or {}).get(key) + + def raw_prev(key: str) -> Optional[float]: + return (prev.get("raw_yuan") or {}).get(key) + + def ratio_of(key: str, values: dict) -> Optional[float]: + rev = values.get("revenue") + v = values.get(key) + return round(v / rev * 100, 2) if (v is not None and rev) else None + + items = [ + {"name": "营业收入", "line_key": "revenue", + "value": cv.get("revenue"), "prev_value": pv.get("revenue"), + "change_rate": _calc_chg(cv.get("revenue"), pv.get("revenue")), + "ratio": 100.0 if cv.get("revenue") is not None else None, + "raw_yuan": raw("revenue"), "prev_raw_yuan": raw_prev("revenue"), + "terms": terms_of("revenue", cur), "prev_terms": terms_of("revenue", prev), + "unit": UNIT, "basis": "按 entity 口径的收入科目贷方发生额（e1=5001 / e2=6001）"}, + {"name": "减：变动成本", "line_key": "variable_cost", + "value": cv.get("variable_cost"), "prev_value": pv.get("variable_cost"), + "change_rate": _calc_chg(cv.get("variable_cost"), pv.get("variable_cost")), + "ratio": ratio_of("variable_cost", cv), + "raw_yuan": raw("variable_cost"), "prev_raw_yuan": raw_prev("variable_cost"), + "terms": terms_of("variable_cost", cur), "prev_terms": terms_of("variable_cost", prev), + "unit": UNIT, "basis": "ENTITY_CALIBERS[entity]['varcost_terms']（带号科目项）"}, + {"name": "＝ 边际贡献", "line_key": "contribution", "is_subtotal": True, + "value": cv.get("contribution"), "prev_value": pv.get("contribution"), + "change_rate": _calc_chg(cv.get("contribution"), pv.get("contribution")), + "ratio": ratio_of("contribution", cv), + "raw_yuan": raw("contribution"), "prev_raw_yuan": raw_prev("contribution"), + "derived_from": ["revenue", "variable_cost"], "unit": UNIT, + "basis": "＝ 营业收入 − 变动成本（元级相减后落万元）"}, + {"name": "减：固定成本", "line_key": "fixed_cost", + "value": cv.get("fixed_cost"), "prev_value": pv.get("fixed_cost"), + "change_rate": _calc_chg(cv.get("fixed_cost"), pv.get("fixed_cost")), + "ratio": ratio_of("fixed_cost", cv), + "raw_yuan": raw("fixed_cost"), "prev_raw_yuan": raw_prev("fixed_cost"), + "terms": terms_of("fixed_cost", cur), "prev_terms": terms_of("fixed_cost", prev), + "unit": UNIT, "basis": "ENTITY_CALIBERS[entity]['fixed_terms'|'fixed_filters']（借−贷净额）"}, + {"name": f"＝ {bottom_label}", "line_key": "operating_profit", "is_total": True, + "value": cv.get("result"), "prev_value": pv.get("result"), + "change_rate": _calc_chg(cv.get("result"), pv.get("result")), + "ratio": ratio_of("result", cv), + "raw_yuan": raw("result"), "prev_raw_yuan": raw_prev("result"), + "derived_from": ["revenue", "variable_cost", "fixed_cost"], "unit": UNIT, + "label": bottom_label, "label_basis": BOTTOM_LABEL_NOTE, + "basis": RECON_FORMULA}, + ] + return items diff --git a/backend/tests/test_profit_summary_real_caliber_20261014.py b/backend/tests/test_profit_summary_real_caliber_20261014.py new file mode 100644 index 00000000..0570e08d --- /dev/null +++ b/backend/tests/test_profit_summary_real_caliber_20261014.py @@ -0,0 +1,398 @@ +"""管理利润表「真实科目取数」重建单测 — project-20261014-cma-profit-summary-real-data-rebuild（2026-10-14） + +背景（代码级审核实锤的 3 处硬伤，本文件逐条设负对照）： + ① 原 `reports.py` L74 `contribution_margin = gross_profit`（边际贡献=毛利）； + ② 原 L76 变动成本 = 营收 × 硬编码比例（五成）； + ③ 原末行标签「＝ 息税前利润」却装净利率算出的 net_profit；且纵向勾稽对不上。 + +断言分组： + L1 逐行可对到真实科目 —— 每行 = 该 entity 口径下真实科目发生额（`rows.parts.by_subject` 留痕） + L2 无硬编码比例 / 无 KPI 比率倒推（AST+源码级负对照）、标签与算法一致、勾稽自检生效（含不平负对照） + L3 按 entity 隔离 —— e1（酣客）/e2（博海）各读自己的科目集；口径改动只改 ENTITY_CALIBERS 即自动跟随 +""" +import ast +import datetime +import os +import re +import sys + +import pytest +from fastapi.testclient import TestClient +from sqlalchemy.orm import Session + +from tests.conftest import create_test_user, get_token_for_user, auth_header + +from app.models import KPIDefinition, KPIValue, VoucherDetail +from app.utils import profit_lines as PL +from app.utils.bep_caliber import ENTITY_CALIBERS + +PERIOD = "2026-09" +PREV = "2026-08" +D = datetime.datetime(2026, 9, 30) + + +# ── 夹具：真实凭证明细（元；行域 = carry_forward==0）──────────────────────── + +def _row(db, entity_id, no, code, debit=0.0, credit=0.0, cf=0, + summary="单测明细", period=PERIOD): + db.add(VoucherDetail(entity_id=entity_id, voucher_no=str(no), voucher_date=D, + subject_code=code, subject_name="单测科目", + debit_amount=debit, credit_amount=credit, summary=summary, + carry_forward=cf, period=period, batch="unit-test")) + db.flush() + + +def seed_e1(db: Session, period: str = PERIOD): + """酣客(e1)口径真账行： + 收入 5001 贷 2,634,900； + 变动 = 5401借 2,000,000 + 5403借 50,000 + 571103借 30,000 − 530103贷 10,000 = 2,070,000； + 固定 = 5601借 120,000 + (5602借 130,000 − 贷 10,000) + 5603借 7,000 = 247,000。 + + 一条 carry_forward=1 结转腿（借 9,999,999）⇒ 任何口径都不得计入。 + """ + _row(db, 1, 1, "5001", credit=2_634_900.0, period=period) + _row(db, 1, 2, "5401", debit=2_000_000.0, period=period) + _row(db, 1, 3, "5403", debit=50_000.0, period=period) + _row(db, 1, 4, "571103", debit=30_000.0, period=period) + _row(db, 1, 5, "530103", credit=10_000.0, period=period) + _row(db, 1, 6, "5601", debit=120_000.0, period=period) # 5601% 前缀 + _row(db, 1, 7, "5602", debit=130_000.0, period=period) + _row(db, 1, 8, "5602", credit=10_000.0, period=period) # 净额 = 120,000 + _row(db, 1, 9, "5603", debit=7_000.0, period=period) + _row(db, 1, 10, "5401", debit=9_999_999.0, cf=1, period=period) # 结转腿，剔除 + db.commit() + + +def seed_e2(db: Session, period: str = PERIOD): + """博海口径真账行： + 收入 6001 贷 1,000,000；变动 = 6401借 800,000； + 固定 = 6601借 60,000 + (6602借 30,000 − 贷 5,000) + (6603借 8,000 − 贷 20,000) = 73,000（净额口径）。 + + 一条 summary='结转期间损溢' 的结转腿（e2 账套 carry_forward 恒 0，必须按 summary 显式剔除）。 + """ + _row(db, 2, 1, "6001", credit=1_000_000.0, period=period) + _row(db, 2, 2, "6401", debit=800_000.0, period=period) + _row(db, 2, 3, "6601", debit=60_000.0, period=period) + _row(db, 2, 4, "6602", debit=30_000.0, period=period) + _row(db, 2, 5, "6602", credit=5_000.0, period=period) + _row(db, 2, 6, "6603", debit=8_000.0, period=period) + _row(db, 2, 7, "6603", credit=20_000.0, period=period) + _row(db, 2, 8, "6602", debit=999_999.0, summary="结转期间损溢", period=period) + db.commit() + + +# ── L1 逐行可对到真实科目 ─────────────────────────────────────────────────── + +class TestLinesFromRealSubjects: + def test_e1_lines_match_real_accounts(self, db: Session): + seed_e1(db) + r = PL.compute_lines(db, 1, PERIOD, PERIOD) + assert r["available"] is True + v = r["values"] + # 万元（元 ÷ 10000；汇总后舍入） + assert v["revenue"] == 263.49, v + assert v["variable_cost"] == 207.00, v + assert v["contribution"] == 56.49, v + assert v["fixed_cost"] == 24.70, v + assert v["result"] == 31.79, v + # 元级留痕：逐项对到真实科目 + assert r["raw_yuan"] == {"revenue": 2_634_900.0, "variable_cost": 2_070_000.0, + "contribution": 564_900.0, "fixed_cost": 247_000.0, + "result": 317_900.0}, r["raw_yuan"] + parts = {p["code"]: p for p in r["rows"]["variable_cost"]["parts"]} + assert parts["5401"]["by_subject"] == {"5401": 2_000_000.0} + assert parts["5403"]["by_subject"] == {"5403": 50_000.0} + assert parts["571103"]["by_subject"] == {"571103": 30_000.0} + # 减项：厂补（贷方）按口径相减 + assert parts["530103"]["sign"] == "-" and parts["530103"]["total_yuan"] == 10_000.0 + # 固定成本：5601% 前缀 + 净额 + fparts = {p["code"]: p for p in r["rows"]["fixed_cost"]["parts"]} + assert fparts["5602%"]["match"] == "like" + assert fparts["5602%"]["total_yuan"] == 120_000.0 # 借 130,000 − 贷 10,000 + assert fparts["5601%"]["total_yuan"] == 120_000.0 + + def test_e1_carry_forward_rows_excluded(self, db: Session): + seed_e1(db) + r = PL.compute_lines(db, 1, PERIOD, PERIOD) + # 结转腿（借 9,999,999，cf=1）不得进 5401 + assert r["rows"]["variable_cost"]["parts"][0]["by_subject"] == {"5401": 2_000_000.0} + assert r["values"]["variable_cost"] == 207.00 + + def test_e2_lines_match_real_accounts_and_exclude_closing(self, db: Session): + seed_e2(db) + r = PL.compute_lines(db, 2, PERIOD, PERIOD) + assert r["available"] is True + v = r["values"] + assert v["revenue"] == 100.00, v + assert v["variable_cost"] == 80.00, v + assert v["contribution"] == 20.00, v + assert v["fixed_cost"] == 7.30, v # 60,000 + 25,000 − 12,000 = 73,000 元 + assert v["result"] == 12.70, v + # 结转腿（summary 含「结转」）不得计入 6602 + fparts = {p["code"]: p for p in r["rows"]["fixed_cost"]["parts"]} + assert fparts["6602%"]["total_yuan"] == 25_000.0, fparts["6602%"] + assert r["caliber"]["exclude_closing"] is True + + +# ── L3 按 entity 隔离 ────────────────────────────────────────────────────── + +class TestEntityIsolation: + def test_e1_and_e2_read_own_subject_sets(self, db: Session): + seed_e1(db) + seed_e2(db) + r1 = PL.compute_lines(db, 1, PERIOD, PERIOD) + r2 = PL.compute_lines(db, 2, PERIOD, PERIOD) + + # 收入基数不同：e1 只算 5001，e2 只算 6001（互不串账） + assert r1["raw_yuan"]["revenue"] == 2_634_900.0 + assert r2["raw_yuan"]["revenue"] == 1_000_000.0 + + # 变动成本科目集不同：e1 含 5401+5403+571103−530103；e2 只有 6401 + assert {p["code"] for p in r1["rows"]["variable_cost"]["parts"]} == {"5401", "5403", "571103", "530103"} + assert {p["code"] for p in r2["rows"]["variable_cost"]["parts"]} == {"6401"} + # e2 的 6401 不得被算进 e1 + assert r1["rows"]["variable_cost"]["parts"][0]["by_subject"] == {"5401": 2_000_000.0} + + # 固定成本科目集不同：e1=5601/5602/5603；e2=6601/6602/6603 + assert {p["code"] for p in r1["rows"]["fixed_cost"]["parts"]} == {"5601%", "5602%", "5603%"} + assert {p["code"] for p in r2["rows"]["fixed_cost"]["parts"]} == {"6601%", "6602%", "6603%"} + + # 口径快照不同（digest 异：口径表按账套而异） + assert r1["caliber"]["digest"] != r2["caliber"]["digest"] + assert r1["caliber"]["name"] == "陕西酣客" and r2["caliber"]["name"] == "博海" + + def test_unregistered_entity_is_marked_missing_not_fabricated(self, db: Session): + """未注册账套（entity=9）⇒ 如实标缺（None + missing），禁跨账套套用口径。""" + seed_e1(db) + r = PL.compute_lines(db, 9, PERIOD, PERIOD) + assert r["available"] is False + assert r["values"]["revenue"] is None and r["values"]["result"] is None + assert r["missing"] and "未在 ENTITY_CALIBERS 注册" in r["missing"][0] + + def test_empty_period_is_marked_missing(self, db: Session): + """该账套该期无明细行 ⇒ 如实标缺，而不是当成 0。""" + seed_e1(db, period="2026-07") + r = PL.compute_lines(db, 1, PERIOD, PERIOD) + assert r["available"] is False and r["values"]["revenue"] is None + assert "无凭证明细行" in r["missing"][0] + + +# ── 口径配置化（口径只有一处：ENTITY_CALIBERS）───────────────────────────── + +class TestCaliberDriven: + def test_fixed_cost_terms_modes(self): + # e1/e2 现行形态 = fixed_filters（前缀 + 借−贷净额）派生 + terms, mode = PL.fixed_cost_terms(ENTITY_CALIBERS[1]) + assert mode == "fixed_filters(net)" + assert terms == [("+", "5601%", "net"), ("+", "5602%", "net"), ("+", "5603%", "net")] + # 口径升级为 fixed_terms（ASK-20261014-04 4-1 形态）时本层零改动跟随 + terms2, mode2 = PL.fixed_cost_terms(dict(ENTITY_CALIBERS[2], + fixed_terms=[("+", "6601", "debit"), ("+", "6603", "debit")])) + assert mode2 == "fixed_terms" + assert terms2 == [("+", "6601", "debit"), ("+", "6603", "debit")] + + def test_caliber_change_flows_through_without_touching_report(self, db: Session, monkeypatch): + """口径改动只改 ENTITY_CALIBERS 一处 ⇒ 取数自动跟随（证明科目集未被写死在报表层）。""" + seed_e1(db) + base = PL.compute_lines(db, 1, PERIOD, PERIOD)["values"]["fixed_cost"] + assert base == 24.70 + # 造一个「口径变更」：固定成本只认 5603% + changed = dict(ENTITY_CALIBERS[1], fixed_filters=[("like", "5603")]) + monkeypatch.setitem(ENTITY_CALIBERS, 1, changed) + after = PL.compute_lines(db, 1, PERIOD, PERIOD)["values"]["fixed_cost"] + assert after == 0.70, after + # 变动成本同样跟随（口径里去掉 5403） + changed2 = dict(ENTITY_CALIBERS[1], varcost_terms=[("+", "5401", "debit")]) + monkeypatch.setitem(ENTITY_CALIBERS, 1, changed2) + assert PL.compute_lines(db, 1, PERIOD, PERIOD)["values"]["variable_cost"] == 200.00 + + def test_aggregation_parity_with_dashboard_and_deriver(self, db: Session): + """取数机制与既有读口/派生口同口径（同夹具同结果 ⇒ 无三套算法漂移）。""" + seed_e1(db) + from app.api.dashboard import _bep_sum_field + from scripts import kpi_bep_deriver as D + + mine = PL.sum_field(db, 1, PERIOD, PERIOD, "net", ("like", "5602")) + assert mine == _bep_sum_field(db, 1, PERIOD, PERIOD, "net", ("like", "5602")) == \ + D._sum_field(db, 1, PERIOD, PERIOD, "net", ("like", "5602")) == 120_000.0 + assert PL.sum_field(db, 1, PERIOD, PERIOD, "credit_amount", ("eq", "5001")) == 2_634_900.0 + + +# ── L2 勾稽自检 ─────────────────────────────────────────────────────────── + +class TestReconciliation: + def test_balanced_when_identity_holds(self): + r = PL.reconcile(100.0, 60.0, 25.0, 15.0) + assert r["balanced"] is True and r["diff"] == 0.0 and r["components_present"] is True + + def test_unbalanced_is_flagged_not_silent(self): + """负对照：末行被其它来源改写（如旧实现的「净利率×收入」）⇒ 必须显式不平。""" + r = PL.reconcile(100.0, 60.0, 25.0, 10.0) # 末行少 5 万 + assert r["balanced"] is False + assert r["diff"] == 5.0 and r["lhs"] == 15.0 and r["rhs"] == 10.0 + assert "勾稽不平" in r["note"] + + def test_missing_components_not_silently_balanced(self): + r = PL.reconcile(100.0, None, 25.0, 15.0) + assert r["balanced"] is False and r["components_present"] is False + assert r["missing"] == ["variable_cost"] and r["diff"] is None + + def test_rounding_tolerance_only(self): + """2 位舍入抖动（≤0.02 万）不误报；真不平（0.5 万）必报。""" + assert PL.reconcile(100.01, 60.0, 25.0, 15.01)["balanced"] is True + assert PL.reconcile(100.0, 60.0, 25.0, 14.5)["balanced"] is False + + +# ── L2 无硬编码比例 / 无 KPI 比率倒推（源码级负对照）─────────────────────── + +def _fn_ast_source(module_file: str, fn_name: str) -> str: + """取函数源码（剥离 docstring，保留代码）——供「无比例常量/无 KPI 比率」断言。""" + tree = ast.parse(open(module_file, encoding="utf-8").read()) + for node in ast.walk(tree): + if isinstance(node, ast.FunctionDef) and node.name == fn_name: + body = node.body + if (body and isinstance(body[0], ast.Expr) + and isinstance(getattr(body[0], "value", None), ast.Constant) + and isinstance(body[0].value.value, str)): + body[0].value.value = "" # 剥离 docstring（口径说明可含科目码/比例字样） + return ast.unparse(node) + raise AssertionError(f"{fn_name} 未找到于 {module_file}") + + +def _string_literals(module_file: str): + """模块内全部字符串字面量（剥离 docstring）——供「无写死科目码」断言。""" + tree = ast.parse(open(module_file, encoding="utf-8").read()) + for node in ast.walk(tree): + if isinstance(node, (ast.Module, ast.ClassDef, ast.FunctionDef, ast.AsyncFunctionDef)): + body = getattr(node, "body", None) or [] + if (body and isinstance(body[0], ast.Expr) + and isinstance(getattr(body[0], "value", None), ast.Constant) + and isinstance(body[0].value.value, str)): + body[0].value.value = "" + return [n.value for n in ast.walk(tree) + if isinstance(n, ast.Constant) and isinstance(n.value, str)] + + +class TestNoHardcodedRatio: + def test_report_endpoint_has_no_ratio_or_kpi_derivation(self): + import app.api.reports as reports_mod + code = _fn_ast_source(reports_mod.__file__, "get_profit_summary") + for bad in ("0.5", "0.50", "* 0.5"): + assert bad not in code, f"硬编码比例残留: {bad}\n{code}" + for bad in ("F_PROFIT_RATE", "F_NET_PROFIT_RATE", "F_COST_RATIO", "F_REVENUE", + "KPIDefinition", "KPIValue"): + assert bad not in code, f"KPI 比率倒推残留: {bad}\n{code}" + # 标签与算法一致：末行标签来自单一常量（＝ 收入 − 变动 − 固定） + assert "BOTTOM_LABEL" in code and "reconcile(" in code + + def test_reports_module_has_no_half_ratio(self): + import app.api.reports as reports_mod + src = open(reports_mod.__file__, encoding="utf-8").read() + assert "0.5" not in src, "reports.py 仍存在 0.5 字面量" + + def test_take_data_module_has_no_ratio_and_no_subject_codes(self): + """取数层：既无比例常量（含 0.5），也无写死的科目码（科目集只能来自 ENTITY_CALIBERS）。""" + src = open(PL.__file__, encoding="utf-8").read() + assert "0.5" not in src, "profit_lines.py 存在 0.5 字面量" + lits = _string_literals(PL.__file__) + offenders = [s for s in lits if re.fullmatch(r"\d{4,}", s or "")] + assert offenders == [], f"profit_lines.py 写死了科目码: {offenders}" + # 口径字段名必须来自 ENTITY_CALIBERS（无自建科目集） + assert "caliber_or_none" in src + for own_key in ("revenue_codes", "variable_cost_codes", "fixed_cost_codes"): + assert own_key not in src, f"自建科目集字段出现: {own_key}" + + +# ── 端点集成（HTTP）─────────────────────────────────────────────────────── + +BASE = "/api/cma/reports/profit-summary" + + +class TestEndpoint: + def test_reconciliation_label_unit_and_change_rate(self, client: TestClient, db: Session): + create_test_user(db) + token = get_token_for_user(client) + seed_e1(db) + seed_e1(db, period=PREV) + # 上期收入改为 100.00 万，便于验证环比 + for row in db.query(VoucherDetail).filter(VoucherDetail.period == PREV, + VoucherDetail.subject_code == "5001").all(): + row.credit_amount = 1_000_000.0 + db.commit() + + resp = client.get(f"{BASE}?period={PERIOD}", headers=auth_header(token)) + assert resp.status_code == 200, resp.text + d = resp.json() + + assert d["unit"] == "万元" and d["bottom_label"] == "营业利润" + assert d["available"] is True and d["missing"] == [] + assert len(d["items"]) == 5 + + items = {i["name"]: i for i in d["items"]} + assert items["营业收入"]["value"] == 263.49 + assert items["营业收入"]["prev_value"] == 100.00 + assert items["营业收入"]["change_rate"] == 163.49 + assert items["减：变动成本"]["value"] == 207.00 + assert items["＝ 边际贡献"]["value"] == 56.49 and items["＝ 边际贡献"]["is_subtotal"] is True + assert items["减：固定成本"]["value"] == 24.70 + # ③ 标签与算法一致：末行名 = 常量；值 = 收入 − 变动 − 固定；无「息税前利润」标签残留 + assert f"＝ {d['bottom_label']}" in items + assert "＝ 息税前利润" not in items + bottom = items[f"＝ {d['bottom_label']}"] + assert bottom["is_total"] is True + assert bottom["value"] == 31.79 + assert bottom["value"] == round(d["reconciliation"]["lhs"], 2) + assert "不标「净利润」" in bottom["label_basis"] + + # ④ 勾稽自检：平衡 + 明细在场 + rec = d["reconciliation"] + assert rec["balanced"] is True and rec["diff"] == 0.0 + assert rec["formula"] == "营业收入 − 变动成本 − 固定成本 = 末行" + + # L1 逐行可对到真实科目（响应体自带 parts/by_subject 取数留痕） + vparts = items["减：变动成本"]["terms"] + assert {p["code"] for p in vparts} == {"5401", "5403", "571103", "530103"} + assert d["caliber"]["digest"] and d["caliber"]["entity_id"] == 1 + assert sum(p["by_subject"].get("5401", 0) for p in vparts) == 2_000_000.0 + + def test_no_kpi_ratio_fallback_when_no_ledger_rows(self, client: TestClient, db: Session): + """负对照：只塞 KPI（旧实现的输入）而账上无行 ⇒ 各行必须标缺，不得按比率倒推。""" + create_test_user(db) + token = get_token_for_user(client) + for code, val in (("F_REVENUE", 100.0), ("F_PROFIT_RATE", 20.0), + ("F_NET_PROFIT_RATE", 10.0), ("F_COST_RATIO", 60.0)): + k = KPIDefinition(kpi_code=code, kpi_name=code, dimension="finance", + target_value=1.0, unit="万元", status="active", + frequency="monthly", entity_id=1) + db.add(k) + db.flush() + db.add(KPIValue(kpi_id=k.id, period=PERIOD, actual_value=val, + source_type="test", source_batch="unit-test", + data_status="verified", entity_id=1)) + db.commit() + + resp = client.get(f"{BASE}?period={PERIOD}", headers=auth_header(token)) + assert resp.status_code == 200 + d = resp.json() + assert d["available"] is False and "无凭证明细行" in d["missing"][0] + assert [i["value"] for i in d["items"]] == [None] * 5 + assert d["reconciliation"]["balanced"] is False # 缺数显式不平，不静默 + assert d["reconciliation"]["components_present"] is False + + def test_entity2_endpoint_isolated_from_entity1_rows(self, client: TestClient, db: Session): + """e2 端点只读 e2 科目集（token 走 entity1 ⇒ 仍读到 e1；helper 层 e2 值不受 e1 行影响）。""" + create_test_user(db) + token = get_token_for_user(client) + seed_e1(db) + seed_e2(db) + r2 = PL.compute_lines(db, 2, PERIOD, PERIOD) + assert r2["values"]["revenue"] == 100.00 and r2["values"]["fixed_cost"] == 7.30 + # entity1 的 token ⇒ entity1 账套（e2 行不进入） + d = client.get(f"{BASE}?period={PERIOD}", headers=auth_header(token)).json() + assert d["entity_id"] == 1 + items = {i["name"]: i for i in d["items"]} + assert items["营业收入"]["value"] == 263.49 + assert {p["code"] for p in items["减：固定成本"]["terms"]} == {"5601%", "5602%", "5603%"} + + +def test_prev_period_helper(): + assert PL.prev_period_of("2026-01") == "2025-12" + assert PL.prev_period_of("2026-10") == "2026-09" diff --git a/backend/tests/test_reports.py b/backend/tests/test_reports.py index cbc94f87..8b79bd69 100644 --- a/backend/tests/test_reports.py +++ b/backend/tests/test_reports.py @@ -91,31 +91,46 @@ class TestProfitSummary: BASE = "/api/cma/reports/profit-summary" def test_profit_summary_with_data(self, client: TestClient, db: Session): - """有数据时：营收/变动成本/边际贡献/固定成本/息税前利润 + 环比""" + """有真账数据时：收入/变动成本/边际贡献/固定成本/营业利润 + 勾稽自检 + + 2026-10-14 重建：原断言编码了被审计实锤的 3 处硬伤（KPI 比率倒推 / 变动成本=营收×五成 / + 末行装净利率值却贴「息税前利润」），故改为断言**真实科目发生额**与勾稽自检。 + 口径与逐行取数证据见 tests/test_profit_summary_real_caliber_20261014.py。 + """ + import datetime as _dt + from app.models import VoucherDetail + create_test_user(db) token = get_token_for_user(client) - seed_profit_kpis(db) + # e1 口径真账：收入 5001 贷 1,000,000；变动 5401 借 600,000；固定 5602 借 250,000（元） + d = _dt.datetime(2026, 6, 30) + for i, (code, dr, cr) in enumerate( + [("5001", 0.0, 1_000_000.0), ("5401", 600_000.0, 0.0), ("5602", 250_000.0, 0.0)], start=1): + db.add(VoucherDetail(entity_id=1, voucher_no=f"U{i}", voucher_date=d, + subject_code=code, subject_name="单测科目", + debit_amount=dr, credit_amount=cr, summary="单测", + carry_forward=0, period="2026-06", batch="unit-test")) + db.commit() resp = client.get(f"{self.BASE}?period=2026-06", headers=auth_header(token)) assert resp.status_code == 200 data = resp.json() assert data["period"] == "2026-06" assert data["prev_period"] == "2026-05" + assert data["available"] is True items = {i["name"]: i for i in data["items"]} - # 营业收入 100 vs 上期80 → +25% assert items["营业收入"]["value"] == 100.0 - assert items["营业收入"]["prev_value"] == 80.0 - assert items["营业收入"]["change_rate"] == 25.0 - # 变动成本 = 营收×50% = 50 - assert items["减：变动成本"]["value"] == 50.0 - # 边际贡献 = 毛利 = 100×20% = 20 - assert items["＝ 边际贡献"]["value"] == 20.0 + assert items["减：变动成本"]["value"] == 60.0 # 5401 借 600,000 元 + assert items["＝ 边际贡献"]["value"] == 40.0 assert items["＝ 边际贡献"]["is_subtotal"] is True - # 固定成本 = 总成本60 - 变动成本50 = 10 - assert items["减：固定成本"]["value"] == 10.0 - # 息税前利润 = 100×10% = 10 - assert items["＝ 息税前利润"]["value"] == 10.0 - assert items["＝ 息税前利润"]["is_total"] is True + assert items["减：固定成本"]["value"] == 25.0 + # 末行标签与算法一致（原为「＝ 息税前利润」装净利率值） + assert data["bottom_label"] == "营业利润" + assert items["＝ 营业利润"]["value"] == 15.0 + assert items["＝ 营业利润"]["is_total"] is True + # 勾稽自检：收入 − 变动成本 − 固定成本 = 末行 + assert data["reconciliation"]["balanced"] is True + assert data["reconciliation"]["diff"] == 0.0 def test_profit_summary_empty_db(self, client: TestClient, db: Session): """无任何KPI数据 → 200，值全部为None，不报错"""