fix(reports): profit-summary 真实科目取数重建(去 KPI 比率倒推/硬编码 0.50 + 勾稽自检 + entity 隔离)
ci/woodpecker/push/main Pipeline was canceled

This commit is contained in:
hermes
2026-10-14 23:14:41 +08:00
parent 6d8ef0d443
commit 2244cd79b7
4 changed files with 829 additions and 118 deletions
+371
View File
@@ -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