feat(cma-report): 资产负债表取数层重分类 R1~R12(调表不调账)
ci/woodpecker/push/main Pipeline failed

依据《CMA 资产负债表重分类规则表 v1》(wecom-finance,任总 2026-09-16 批准落地)。

新增 backend/app/services/report_reclass.py:12 条规则挂 CMA 真实科目编码的取数层重分类引擎
- 余额方向规范化:资产 sum(借-贷) / 负债·权益 sum(贷-借),正常余额一律为正
- 触发判定全部基于原始余额(单遍、幂等),R3/R4 等互为镜像的规则互不干扰
- 科目取数前缀匹配(1122% 覆盖 112201 明细);备抵科目 1602 反向取数(修正固定资产净额虚增 2 倍折旧)
- 只读 voucher_details + 作废守卫,不生成记账凭证
- 输出:触发条目 / 不变量 / 负数清单 / 模板待核 / 未列报科目 / 损益未结转 诊断

reports.py:
- /balance-sheet 接入重分类;新增 basis 参数(end=期末余额累计为默认 / period=本期发生额兼容回退)
- 有凭证明细时不再混入示例数据(展示口径与落库口径同源);行项目增 raw_value/reclass_adjustment 可追溯
- 规则落位项目(模板未单列)按虚拟行列报;补「长期待摊费用(1801)」行、未分配利润补 3103/3104 别名
- 新增 /balance-sheet/reclass-working-paper(+ /export) 底稿 JSON/Excel;statutory 导出新增「重分类底稿」Sheet

测试 tests/test_report_reclass.py 14 例(触发/不变量/幂等/多租户/口径/底稿/虚拟行/阻塞规则)
全量 pytest:1108 passed, 1 xfailed。现网只读验收 scripts/verify_reclass_live_20260916.py(酣客 entity1 2026-08)
This commit is contained in:
Hermes CI Fix
2026-09-16 00:38:01 +08:00
parent 7b74736980
commit ba85e03f2f
4 changed files with 1333 additions and 8 deletions
+621
View File
@@ -0,0 +1,621 @@
"""资产负债表取数层重分类引擎(R1~R12,调表不调账)
依据:《CMA 资产负债表重分类规则表 v1》
(`wecom-finance/reports/CMA_资产负债表重分类规则表_v1.md`,任总 2026-09-16 批准落地)
本模块只作用于**报表取数层**:`voucher_details` 只读,**不生成记账凭证、不改账簿**
(调表不调账;账实不符走差错更正,见规则表 §四)。
口径要点
--------
1. **余额方向**:资产类 NSB = Σ(借−贷);负债/权益类 NSB = Σ(贷−借)。NSB = normal-side balance。
2. **触发**:NSB < 0(明细科目反方向余额,如预收账款出现借方余额)→ 转正额 = |NSB|,
移入目标报表项目(准则 30 号「不得抵销」)。
3. **全部触发判定基于原始余额**(单遍、顺序无关、可重复执行/幂等)——避免 R3/R4 这类互为镜像
的规则互相影响(如 2202 应收账款 ↔ 1123 预付账款 双向都有反向余额时)。
4. **科目取数一律前缀匹配**(`subject_code LIKE '1122%'`,与规则表 §二 公式一致):CMA 科目为多级
(100101 / 22210101 / 310415),精确匹配会漏掉全部明细行。
5. **basis**:`end` = 期末余额(累计至 period,含期初/结转,资产负债表口径,默认);
`period` = 本期发生额(兼容旧报表口径)。
6. **不变量**:跨方向重分类(资产↔负债)使资产合计与负债合计**同额增加**,
故「总资产不变」只在同侧列报下成立;真正恒定的不变量是
`Δ资产合计 = Δ负债合计` 且 `资产合计 − 负债合计 − 所有者权益合计` 前后不变。
本模块把这两条都算出来(`invariants`),不隐藏 `total_assets_unchanged=False` 这一事实。
"""
from __future__ import annotations
from dataclasses import dataclass, field
from decimal import Decimal
from typing import Dict, List, Optional, Sequence, Tuple
from sqlalchemy import func
from app.utils.voucher_guard import apply_voucher_guard
RULE_VERSION = "reclass-v1.0-20260916"
DEFAULT_TOL = 0.01 # 元;小于该值的反方向余额不触发(浮点/分位噪声)
# 报表板块 → 余额方向
SIDE_OF_SECTION = {
"current_assets": "asset",
"non_current_assets": "asset",
"current_liabilities": "liability",
"non_current_liabilities": "liability",
"equity": "equity",
}
# 资产备抵科目(余额在贷方)→ 取数方向反转:1602 累计折旧 / 1603 固定资产减值准备 / 1604 在建工程减值准备
CONTRA_CODES: Dict[str, str] = {"1602": "liability", "1603": "liability", "1604": "liability"}
@dataclass(frozen=True)
class ReclassRule:
"""一条重分类规则(规则表 §二 的机器可读形式)。"""
code: str # R1~R12
source_code: str # 来源科目编码(前缀匹配)
source_name: str
source_side: str # asset / liability
target_item: str # 目标报表项目名(小企业准则项目池)
target_key: str # 目标项目在报表条目表里的 key
target_side: str # asset / liability
target_codes: Tuple[str, ...] = ()
status: str = "active" # active=可落地 / template_gap=目标项模板未列报(引擎按虚拟行列报)/ blocked=缺数据或准则未列报
error_check: bool = False # 触发后需先排查记账差错(R10)
note: str = ""
# ── R1~R12(规则表 §二,挂 CMA 真实科目编码) ────────────────────────────────
RECLASS_RULES: Tuple[ReclassRule, ...] = (
ReclassRule("R1", "1122", "应收账款", "asset", "预收账款", "2203", "liability",
("2203",), note="小企业准则无「合同负债」,目标为预收账款"),
ReclassRule("R2", "2203", "预收账款", "liability", "应收账款", "1122", "asset", ("1122",),
note="R1 反向对称"),
ReclassRule("R3", "2202", "应付账款", "liability", "预付账款", "1123", "asset", ("1123",),
note="应付账款出现借方余额(多付/预付)"),
ReclassRule("R4", "1123", "预付账款", "asset", "应付账款", "2202", "liability", ("2202",),
note="R3 反向对称"),
ReclassRule("R5", "1221", "其他应收款", "asset", "其他应付款", "2241", "liability", ("2241",)),
ReclassRule("R6", "2241", "其他应付款", "liability", "其他应收款", "1221", "asset", ("1221",)),
ReclassRule("R7", "1121", "应收票据", "asset", "应付票据", "2201", "liability", ("2201",),
status="template_gap", note="CMA 报表模板未单列应收/应付票据 → 引擎按虚拟行列报,待财务确认"),
ReclassRule("R8", "2201", "应付票据", "liability", "应收票据", "1121", "asset", ("1121",),
status="template_gap", note="CMA 报表模板未单列应收/应付票据 → 引擎按虚拟行列报,待财务确认"),
ReclassRule("R9", "2221", "应交税费", "liability", "其他流动资产", "OCA_OTHER", "asset",
status="template_gap",
note="留抵/预缴/多缴(借方余额);≤1年→其他流动资产,>1年→其他非流动资产。"
"CMA 无到期日字段,暂统一列报「其他流动资产」并标注待核"),
ReclassRule("R10", "2211", "应付职工薪酬", "liability", "其他应收款", "1221", "asset", ("1221",),
error_check=True, note="先查是否记账差错(多付/重复发放),差错走更正、非差错才重分类"),
ReclassRule("R11", "2501", "长期借款/应付债券", "liability", "一年内到期的非流动负债",
"NCL_DUE_1Y", "liability", status="blocked",
note="小企业准则报表未单列「一年内到期的非流动负债」,且 voucher_details 无到期日字段 → 不落地"),
ReclassRule("R12", "1501", "长期债权投资/长期应收款", "asset", "一年内到期的非流动资产",
"NCA_DUE_1Y", "asset", status="blocked",
note="小企业准则报表未单列「一年内到期的非流动资产」,且无到期日字段 → 不落地"),
)
# ── 模板未列报、但规则需要落位的报表项目(虚拟行;仅当金额非零时列报) ──────────
EXTRA_ITEMS: Dict[str, dict] = {
"1121": {"name": "应收票据", "section_key": "current_assets", "side": "asset", "codes": [("1121", 1)],
"template_gap": True, "note": "小企业准则项目池有「应收票据」,CMA 模板未列报"},
"2201": {"name": "应付票据", "section_key": "current_liabilities", "side": "liability", "codes": [("2201", 1)],
"template_gap": True, "note": "小企业准则项目池有「应付票据」,CMA 模板未列报"},
"OCA_OTHER": {"name": "其他流动资产", "section_key": "current_assets", "side": "asset", "codes": [],
"template_gap": True, "ns_category": "operating",
"note": "小企业准则项目池有「其他流动资产」,CMA 模板未列报(R9 落位项目)"},
"ONCA_OTHER": {"name": "其他非流动资产", "section_key": "non_current_assets", "side": "asset", "codes": [],
"template_gap": True, "blocked": True, "note": "R9 的 >1年落位项目;无到期日字段,暂不落地"},
"OCL_OTHER": {"name": "其他流动负债", "section_key": "current_liabilities", "side": "liability", "codes": [],
"template_gap": True, "blocked": True, "note": "R7 备选落位项目;暂不落地"},
"NCA_DUE_1Y": {"name": "一年内到期的非流动资产", "section_key": "current_assets", "side": "asset", "codes": [],
"template_gap": True, "blocked": True, "note": "小企业准则未单列(R12 待核)"},
"NCL_DUE_1Y": {"name": "一年内到期的非流动负债", "section_key": "current_liabilities", "side": "liability", "codes": [],
"template_gap": True, "blocked": True, "note": "小企业准则未单列(R11 待核)"},
}
# ── 余额取数(前缀匹配 + 作废守卫 + basis 口径) ─────────────────────────────
class BalanceSource:
"""按 (品类, 期间, basis) 取科目 normal-side balance(NSB),带会话内缓存。"""
def __init__(self, db, entity_id: int, period: str, basis: str = "end", tol: float = DEFAULT_TOL):
self.db = db
self.entity_id = entity_id
self.period = period
self.basis = "period" if basis == "period" else "end"
self.tol = tol
self._cache: Dict[Tuple[str, str], Optional[Decimal]] = {}
self._has_data: Optional[bool] = None
# -- 基础查询 ---------------------------------------------------------
def _query(self, code: str, period: str):
from app.models import VoucherDetail
q = self.db.query(
func.sum(VoucherDetail.debit_amount),
func.sum(VoucherDetail.credit_amount),
func.count(VoucherDetail.id),
).filter(
VoucherDetail.entity_id == self.entity_id,
VoucherDetail.subject_code.like(f"{code}%"),
)
if self.basis == "end":
q = q.filter(VoucherDetail.period <= period) # 期末余额 = 累计(含期初/结转)
else:
q = q.filter(VoucherDetail.period == period) # 本期发生额
return apply_voucher_guard(self.db, q, VoucherDetail)
def nsb(self, code: str, side: str, period: Optional[str] = None) -> Optional[Decimal]:
"""NSB = 资产 Σ(借−贷) / 负债·权益 Σ(贷−借);无数据返回 None。"""
period = period or self.period
ck = (f"{code}|{side}|{period}|{self.basis}", "")
if ck in self._cache:
return self._cache[ck]
d = c = n = None
try:
row = self._query(code, period).one()
d, c, n = row[0], row[1], row[2]
except Exception: # noqa: BLE001 - 无凭证明细表/查询异常 → 视为无数据
pass
if not n:
self._cache[ck] = None
return None
debit = Decimal(str(d or 0))
credit = Decimal(str(c or 0))
val = (debit - credit) if side == "asset" else (credit - debit)
self._cache[ck] = val
return val
def has_data(self) -> bool:
if self._has_data is None:
from app.models import VoucherDetail
q = self.db.query(func.count(VoucherDetail.id)).filter(
VoucherDetail.entity_id == self.entity_id
)
if self.basis == "period":
q = q.filter(VoucherDetail.period == self.period)
try:
self._has_data = bool(q.scalar())
except Exception: # noqa: BLE001
self._has_data = False
return bool(self._has_data)
# ── 报表条目表(模板行 + 规则落位项) ──────────────────────────────────────
def build_items(template_sections: Sequence[dict]) -> Dict[str, dict]:
"""从报表模板派生条目表;key 与模板行 key 一致("|".join(codes))。
`alias_codes` 只参与取数、不参与 key(例如模板用 4103/4104、数据用 3103/3104)。
"""
items: Dict[str, dict] = {}
for sec in template_sections or []:
side = SIDE_OF_SECTION.get(sec["key"], "asset")
for line in sec.get("lines", []):
key = "|".join(c for c, _ in line["codes"])
codes = list(line["codes"]) + list(line.get("alias_codes") or [])
items[key] = {
"key": key,
"name": line["name"],
"section_key": sec["key"],
"side": side,
"codes": codes,
"template": True,
"template_gap": False,
"ns_category": line.get("ns_category"),
"aliased": bool(line.get("alias_codes")),
}
for key, cfg in EXTRA_ITEMS.items():
if key in items:
continue
items[key] = {
"key": key, "name": cfg["name"], "section_key": cfg["section_key"],
"side": cfg["side"], "codes": list(cfg.get("codes") or []),
"template": False, "template_gap": bool(cfg.get("template_gap")),
"blocked": bool(cfg.get("blocked")), "ns_category": cfg.get("ns_category"),
"note": cfg.get("note", ""),
}
return items
def _item_raw(bs: BalanceSource, item: dict, period: Optional[str] = None) -> Optional[Decimal]:
"""条目原始余额:Σ sign × NSB(code);无任何数据返回 None(纯落位项目按 0 处理)。
备抵科目(累计折旧等)余额方向与所属板块相反,取数时反向(`CONTRA_CODES`)——
否则「固定资产净额 = 原值 − 累计折旧」会被算成 原值 + 累计折旧。
"""
total = Decimal(0)
has = False
for code, sign in item["codes"]:
side = CONTRA_CODES.get(code) or item["side"]
v = bs.nsb(code, side, period)
if v is not None:
total += v * sign
has = True
if not has:
return Decimal(0) if (not item["template"] or not item["codes"]) else None
return total
def _covered_code_map(items: Dict[str, dict]) -> Dict[str, str]:
"""科目编码 → 条目 key(取数代码 sign=1 的;用于判定规则来源项落位)。"""
m: Dict[str, str] = {}
for key, item in items.items():
for code, sign in item["codes"]:
if sign == 1:
m.setdefault(code, key)
return m
# ── 主入口:一次性算出「原值 / 重分类后 / 底稿 / 不变量 / 诊断」 ─────────────
def analyze(db, entity_id: int, period: str, basis: str = "end",
template_sections: Optional[Sequence[dict]] = None,
rules: Sequence[ReclassRule] = RECLASS_RULES,
tol: float = DEFAULT_TOL, include_unmapped: bool = True) -> dict:
"""资产负债表取数层重分类分析(不写库、幂等、可重复调用)。"""
bs = BalanceSource(db, entity_id, period, basis, tol)
items = build_items(template_sections or [])
code_to_key = _covered_code_map(items)
raw: Dict[str, Optional[Decimal]] = {}
prev: Dict[str, Optional[Decimal]] = {}
for key, item in items.items():
raw[key] = _item_raw(bs, item)
if basis == "end":
prev[key] = _item_raw(bs, item, _prev_period(period))
# ── 触发判定(全部基于原始余额,单遍) ──
entries: List[dict] = []
for r in rules:
if r.status == "blocked":
continue
v = bs.nsb(r.source_code, r.source_side)
if v is None or v >= -Decimal(str(tol)):
continue
amount = float(-v)
src_key = code_to_key.get(r.source_code, r.source_code)
tgt = items.get(r.target_key) or {}
entries.append({
"rule": r.code,
"source_code": r.source_code,
"source_name": r.source_name,
"source_side": r.source_side,
"source_key": src_key,
"raw_balance": round(float(v), 2),
"abnormal_direction": "贷方余额" if r.source_side == "asset" else "借方余额",
"reclass_amount": round(amount, 2),
"target_item": r.target_item,
"target_key": r.target_key,
"target_codes": list(r.target_codes),
"target_side": r.target_side,
"target_in_template": bool(tgt.get("template")),
"rule_status": r.status,
"error_check": r.error_check,
"note": r.note,
})
# ── 重分类后金额 ──
after: Dict[str, Optional[float]] = {}
for key, v in raw.items():
after[key] = float(v) if v is not None else 0.0
for e in entries:
after[e["source_key"]] = float(after.get(e["source_key"]) or 0.0) + e["reclass_amount"]
after[e["target_key"]] = float(after.get(e["target_key"]) or 0.0) + e["reclass_amount"]
def _totals(values: Dict[str, Optional[float]]) -> dict:
t = {"assets": 0.0, "liabilities": 0.0, "equity": 0.0}
for key, item in items.items():
v = values.get(key)
if v is None or item.get("blocked"):
continue
if item["side"] == "asset":
t["assets"] += float(v)
elif item["side"] == "liability":
t["liabilities"] += float(v)
else:
t["equity"] += float(v)
t["liab_equity"] = t["liabilities"] + t["equity"]
t["residual"] = round(t["assets"] - t["liab_equity"], 2)
return {k: round(v, 2) for k, v in t.items()}
before_t = _totals(raw)
after_t = _totals(after)
reclass_total = round(sum(e["reclass_amount"] for e in entries), 2)
in_scope = set()
for r in rules:
if r.status == "blocked":
continue
in_scope.add(code_to_key.get(r.source_code, r.source_code))
in_scope.add(r.target_key)
def _negatives(values: Dict[str, Optional[float]]) -> List[dict]:
out = []
for key, item in items.items():
v = values.get(key)
if v is None or v >= -tol:
continue
out.append({
"key": key, "name": item["name"], "side": item["side"],
"section_key": item["section_key"], "value": round(float(v), 2),
"in_reclass_scope": key in in_scope,
"template": item["template"],
})
return sorted(out, key=lambda x: x["value"])
neg_before = _negatives(raw)
neg_after = _negatives(after)
neg_after_in_scope = [n for n in neg_after if n["in_reclass_scope"]]
invariants = {
"basis": bs.basis,
"period": period,
"rule_version": RULE_VERSION,
"assets_before": before_t["assets"],
"assets_after": after_t["assets"],
"liabilities_before": before_t["liabilities"],
"liabilities_after": after_t["liabilities"],
"equity_before": before_t["equity"],
"equity_after": after_t["equity"],
"assets_delta": round(after_t["assets"] - before_t["assets"], 2),
"liabilities_delta": round(after_t["liabilities"] - before_t["liabilities"], 2),
"equity_delta": round(after_t["equity"] - before_t["equity"], 2),
"reclass_total": reclass_total,
# ① 调表不调账:跨方向重分类 → 资产/负债同额变动
"delta_assets_equals_delta_liabilities":
abs((after_t["assets"] - before_t["assets"]) - (after_t["liabilities"] - before_t["liabilities"])) < 0.01,
# ② 权益不受影响
"equity_unchanged": abs(after_t["equity"] - before_t["equity"]) < 0.01,
# ③ 勾稽差额不被重分类改变(重分类不制造新的不平)
"residual_before": before_t["residual"],
"residual_after": after_t["residual"],
"residual_unchanged": abs(after_t["residual"] - before_t["residual"]) < 0.01,
# ④ 勾稽恒等式(资产合计 = 负债 + 所有者权益)
"identity_holds_before": abs(before_t["residual"]) < 0.01,
"identity_holds_after": abs(after_t["residual"]) < 0.01,
# ⑤ 「总资产不变」:跨方向重分类下必然 False(见模块 docstring 第 6 条)
"total_assets_unchanged": abs(after_t["assets"] - before_t["assets"]) < 0.01,
"total_assets_unchanged_semantics": (
"跨方向重分类(资产↔负债)使资产合计与负债合计同额变动(Δ资产=Δ负债=转正额);"
"「总资产不变」仅在同侧列报时成立。恒定不变量 = Δ资产=Δ负债 且 资产−负债−权益 不变。"
),
"no_negative_in_scope_after": len(neg_after_in_scope) == 0,
"negatives_before": len(neg_before),
"negatives_after": len(neg_after),
"entries_count": len(entries),
}
template_gaps: List[dict] = []
for r in rules:
tgt = items.get(r.target_key) or {}
if r.status == "active":
continue
template_gaps.append({
"rule": r.code, "source_code": r.source_code, "target_item": r.target_item,
"target_key": r.target_key, "status": r.status,
"in_report": (r.status != "blocked") and bool(tgt.get("template_gap")),
"note": r.note,
})
if not bs.has_data():
entries = []
after = {k: (float(v) if v is not None else None) for k, v in raw.items()}
# 无凭证明细(如 entity2)→ 不做重分类,全部回落到示例数据路径
for t in (before_t, after_t):
t.update({"assets": 0.0, "liabilities": 0.0, "equity": 0.0,
"liab_equity": 0.0, "residual": 0.0})
invariants.update({
"assets_after": 0.0, "liabilities_after": 0.0, "equity_after": 0.0,
"assets_delta": 0.0, "liabilities_delta": 0.0, "equity_delta": 0.0,
"reclass_total": 0.0, "entries_count": 0,
"delta_assets_equals_delta_liabilities": True, "equity_unchanged": True,
"residual_after": 0.0, "residual_unchanged": True,
"identity_holds_after": True, "no_negative_in_scope_after": True,
"negatives_before": 0, "negatives_after": 0,
})
neg_before = neg_after = neg_after_in_scope = []
return {
"rule_version": RULE_VERSION,
"basis": bs.basis,
"period": period,
"entity_id": entity_id,
"prev_period": _prev_period(period),
"has_voucher_data": bs.has_data(),
"rules": [r.__dict__ for r in rules],
"items": items,
"raw": {k: (float(v) if v is not None else None) for k, v in raw.items()},
"after": after,
"prev_raw": {k: (float(v) if v is not None else None) for k, v in prev.items()},
"entries": entries,
"invariants": invariants,
"totals_before": before_t,
"totals_after": after_t,
"negatives": {"before": neg_before, "after": neg_after, "after_in_scope": neg_after_in_scope},
"template_gaps": template_gaps,
"unmapped_subjects": unmapped_subjects(db, entity_id, period, bs.basis, items) if include_unmapped else [],
"pl_unclosed": pl_unclosed(db, entity_id, period, bs.basis) if include_unmapped else None,
}
def _prev_period(period: str) -> str:
"""上一期间 YYYY-MM → YYYY-(MM-1)"""
try:
y, m = period.split("-")
y, m = int(y), int(m)
m -= 1
if m <= 0:
m += 12
y -= 1
return f"{y}-{m:02d}"
except Exception: # noqa: BLE001
return period
def pl_unclosed(db, entity_id: int, period: str, basis: str) -> Optional[float]:
"""损益类科目未结转余额(含结转凭证;口径 贷−借)——正常应为 0。
非 0 即为「勾稽差额」的直接来源(结转不完整),用于把不变量差额解释清楚。
"""
from app.models import VoucherDetail
from sqlalchemy import or_
try:
q = db.query(func.sum(VoucherDetail.credit_amount - VoucherDetail.debit_amount)).filter(
VoucherDetail.entity_id == entity_id,
or_(VoucherDetail.subject_code.like("5%"), VoucherDetail.subject_code.like("6%")),
)
if basis == "end":
q = q.filter(VoucherDetail.period <= period)
else:
q = q.filter(VoucherDetail.period == period)
v = apply_voucher_guard(db, q, VoucherDetail).scalar()
return round(float(v or 0), 2)
except Exception: # noqa: BLE001
return None
def _resolve_category(code, cats: Dict[str, str]) -> Optional[str]:
"""科目 → asset/liability/equity:先精确、再逐级取上级编码(明细科目 category='detail')。"""
s = str(code)
for n in (len(s), 6, 4, 2):
if n > len(s):
continue
c = cats.get(s[:n])
if c in ("asset", "liability", "equity"):
return c
return None
def unmapped_subjects(db, entity_id: int, period: str, basis: str, items: Dict[str, dict]) -> List[dict]:
"""凭证明细里出现、但报表模板/规则都未列报的资产·负债·权益科目余额(勾稽差额诊断)。
只做诊断,不参与重分类——用于定位「资产合计 ≠ 负债 + 权益」的差额来源。
"""
from app.models import VoucherDetail, Subject
try:
q = db.query(
VoucherDetail.subject_code,
func.sum(VoucherDetail.debit_amount - VoucherDetail.credit_amount),
func.count(VoucherDetail.id),
).filter(VoucherDetail.entity_id == entity_id)
if basis == "end":
q = q.filter(VoucherDetail.period <= period)
else:
q = q.filter(VoucherDetail.period == period)
rows = apply_voucher_guard(db, q, VoucherDetail).group_by(VoucherDetail.subject_code).all()
cats = {c: cat for c, cat in db.query(Subject.subject_code, Subject.category).all()}
except Exception: # noqa: BLE001
return []
covered = []
for key, item in items.items():
for code, _ in item["codes"]:
covered.append(code)
agg: Dict[str, dict] = {}
for code, net, n in rows:
cat = _resolve_category(code, cats)
if cat is None:
continue
if any(str(code) == c or str(code).startswith(c) for c in covered if c):
continue
bucket = str(code)[:4] if len(str(code)) > 4 else str(code)
side = "asset" if cat == "asset" else ("liability" if cat == "liability" else "equity")
d = agg.setdefault(bucket, {"subject_code": bucket, "category": cat, "side": side,
"net_debit_minus_credit": 0.0, "rows": 0, "detail_codes": []})
d["net_debit_minus_credit"] += float(net or 0)
d["rows"] += int(n or 0)
if str(code) != bucket:
d["detail_codes"].append(str(code))
out = []
for d in agg.values():
d["net_debit_minus_credit"] = round(d["net_debit_minus_credit"], 2)
d["normal_side_balance"] = round(
d["net_debit_minus_credit"] if d["side"] == "asset" else -d["net_debit_minus_credit"], 2)
out.append(d)
return sorted(out, key=lambda x: -abs(x["normal_side_balance"]))
# ── 底稿(审计可追溯) ─────────────────────────────────────────────────────
def entry_rows(analysis: dict) -> List[dict]:
"""重分类底稿明细行:源科目/余额/转正额/目标项/依据。"""
rows = []
for e in analysis["entries"]:
rows.append({
"规则": e["rule"],
"来源科目": e["source_code"],
"科目名称": e["source_name"],
"原余额(元)": e["raw_balance"],
"异常方向": e["abnormal_direction"],
"转正额(元)": e["reclass_amount"],
"目标报表项目": e["target_item"],
"目标科目": "/".join(e["target_codes"]) or "-",
"目标项模板已列报": "是" if e["target_in_template"] else "否(虚拟行)",
"处理": "调表不调账",
"需先行差错核查": "是" if e["error_check"] else "否",
"依据/备注": e["note"],
})
return rows
def diagnostic_rows(analysis: dict) -> List[dict]:
"""底稿诊断区:不变量 + 负数项目 + 模板未列报 + 未列报科目(勾稽差额诊断)。"""
inv = analysis["invariants"]
rows = [
{"项目": "规则版本", "值": inv["rule_version"], "说明": "取数层重分类规则集"},
{"项目": "取数口径 basis", "值": inv["basis"], "说明": "end=期末余额(累计至期间);period=本期发生额"},
{"项目": "触发规则数", "值": inv["entries_count"], "说明": "本轮实际触发的重分类条数"},
{"项目": "转正额合计(元)", "值": inv["reclass_total"], "说明": "调表不调账,不生成凭证"},
{"项目": "资产合计-重分类前(元)", "值": inv["assets_before"], "说明": ""},
{"项目": "资产合计-重分类后(元)", "值": inv["assets_after"], "说明": ""},
{"项目": "负债合计-重分类前(元)", "值": inv["liabilities_before"], "说明": ""},
{"项目": "负债合计-重分类后(元)", "值": inv["liabilities_after"], "说明": ""},
{"项目": "所有者权益合计-重分类前(元)", "值": inv["equity_before"], "说明": ""},
{"项目": "所有者权益合计-重分类后(元)", "值": inv["equity_after"], "说明": ""},
{"项目": "Δ资产 = Δ负债", "值": "PASS" if inv["delta_assets_equals_delta_liabilities"] else "FAIL",
"说明": f"Δ资产={inv['assets_delta']},Δ负债={inv['liabilities_delta']}"},
{"项目": "权益不变", "值": "PASS" if inv["equity_unchanged"] else "FAIL", "说明": ""},
{"项目": "勾稽差额不变", "值": "PASS" if inv["residual_unchanged"] else "FAIL",
"说明": f"前={inv['residual_before']},后={inv['residual_after']}"},
{"项目": "勾稽恒等式(资产=负债+权益)", "值": "平衡" if inv["identity_holds_after"] else "不平",
"说明": f"差额 {inv['residual_after']} 元(重分类前 {inv['residual_before']})"},
{"项目": "总资产不变", "值": "是" if inv["total_assets_unchanged"] else "否(跨方向重分类必然变化)",
"说明": inv["total_assets_unchanged_semantics"]},
{"项目": "规则范围内负数项目(重分类后)", "值": len(analysis["negatives"]["after_in_scope"]),
"说明": "应为 0"},
{"项目": "负数项目总数(重分类前/后)", "值": f"{inv['negatives_before']} / {inv['negatives_after']}",
"说明": "重分类范围外科目见下方清单"},
]
for n in analysis["negatives"]["after"]:
rows.append({
"项目": f"负数项目:{n['name']}",
"值": n["value"],
"说明": ("规则范围内(异常,需查)" if n["in_reclass_scope"]
else "R1~R12 未覆盖科目 → 差错更正/取数口径问题,需财务判定"),
})
for g in analysis["template_gaps"]:
rows.append({"项目": f"模板待核:{g['rule']} {g['source_code']}",
"值": g["status"], "说明": g["note"]})
plu = analysis.get("pl_unclosed")
if plu is not None:
rows.append({
"项目": "损益类未结转余额(应为 0)",
"值": plu,
"说明": ("与勾稽差额一致 → 差额来自结转不完整,需差错更正(非重分类能解决)"
if abs(plu) > 0.01 and abs(plu - inv.get("residual_after", 0)) < 0.01
else ("非 0,需差错核查" if abs(plu) > 0.01 else "已结平")),
})
for u in analysis["unmapped_subjects"]:
rows.append({
"项目": f"未列报科目:{u['subject_code']}",
"值": u["normal_side_balance"],
"说明": f"{u['side']};明细 {'/'.join(u['detail_codes']) or '-'};模板未列报 → 影响勾稽差额",
})
return rows