sxbh_admin/cma-management
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(cma-report): 建账期初层 + 方向判定可信度 + 权益行科目集（M1/M2/M7，规则表 v1.1）

Browse Source 
- M1 时点报表口径：余额 = 期初(report_opening_balances) + 累计发生额；期初只作用于取数层
  （不生成凭证/不写 voucher_details/不改 kpi_values）；basis=period 不套期初；受 effective_date 约束
- M2 期初未补 → direction_check_unreliable=true：报表/底稿带"期初未补，方向判定不可信"水印，
  重分类触发只进底稿诊断（diagnostic_only），不落地对外行项目
- M7 权益行科目集 = {3001,3002,3101,3103,3104,3105}（含 4001/4002/4101/4103/4104 别名）：补实收资本 3001
- 1901 待处理财产损溢过渡列报「其他流动资产」以保恒等式（M6 归因文案）
- 只读验收脚本 + 幂等导入器（dry-run 默认，--commit 门控）
This commit is contained in:
Hermes CI Fix
2026-09-16 01:17:09 +08:00
parent db3c42f766
commit ae6f71dbc0
7 changed files with 833 additions and 29 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/app/api/reports.py
+46 -10
View File
@@ -1410,12 +1410,19 @@ BALANCE_SHEET_SECTIONS = [
"key": "equity",
"name": "所有者权益",
"category_label": "所有者权益",
# M7（规则表 v1.1 §三）：权益行科目集 = {3001,3002,3101,3103,3104,3105}
# （含 4001/4002/4101/4103/4104 别名）。账套实际记账用小企业准则 3xxx 科目，
# 别名只参与取数、不参与行 key（避免打破按 key 的预算映射）。缺 3001 会漏实收资本 2,000,000。
"lines": [
{"codes": [("4001", 1)], "name": "实收资本", "ns_category": "equity"},
{"codes": [("4002", 1)], "name": "资本公积", "ns_category": "equity"},
{"codes": [("4001", 1)], "alias_codes": [("3001", 1)],
"name": "实收资本", "ns_category": "equity"},
{"codes": [("4002", 1)], "alias_codes": [("3002", 1)],
"name": "资本公积", "ns_category": "equity"},
{"codes": [("4101", 1)], "alias_codes": [("3101", 1)],
"name": "盈余公积", "ns_category": "equity"},
# alias_codes：模板用 4103/4104，智享通凭证明细实际用 3103/3104（2026-09-16 补）
{"codes": [("4103", 1), ("4104", 1)],
"alias_codes": [("3103", 1), ("3104", 1)],
"alias_codes": [("3103", 1), ("3104", 1), ("3105", 1)],
"name": "未分配利润", "ns_category": "equity"},
],
},
@@ -1549,20 +1556,26 @@ def get_balance_sheet(
plan = _bs_analysis(db, entity_id, period, use_basis)
has_real = bool(plan["has_voucher_data"])
# M2：期初未补 → 方向判定不可信 → 重分类结论只进底稿诊断，不落地到对外行项目
diag_only = bool(plan.get("direction_check_unreliable"))
plan_prev = _bs_analysis(db, entity_id, prev_period, use_basis, include_unmapped=False) if has_real else None
def _present_key(pl: Optional[dict]) -> str:
"""M2：期初未补（方向判定不可信）时，对外行项目用**原始余额**（不呈现"重分类转正"）"""
return "raw" if (pl or {}).get("direction_check_unreliable") else "after"
def _line_value(pl: Optional[dict], key: str, col: str):
if pl is None:
return None
return pl["after"].get(key)
return pl[_present_key(pl)].get(key)
# 规则落位项目（模板未列报的目标项）→ 按板块挂载虚拟行
virtual: dict = {}
for key, item in plan["items"].items():
if item["template"] or item.get("blocked"):
continue
end_v = plan["after"].get(key)
begin_v = plan_prev["after"].get(key) if plan_prev else None
end_v = plan[_present_key(plan)].get(key)
begin_v = plan_prev[_present_key(plan_prev)].get(key) if plan_prev else None
if not end_v and not begin_v:
continue
virtual.setdefault(item["section_key"], []).append({
@@ -1604,6 +1617,8 @@ def get_balance_sheet(
if begin["value"] is not None:
sec_begin += begin["value"]
raw_key = plan["raw"].get(key)
diag_adj = None if raw_key is None else round(
float(plan["after"].get(key) or 0) - float(raw_key), 2)
lines.append({
"name": line["name"],
"item_key": key,
@@ -1615,8 +1630,11 @@ def get_balance_sheet(
"is_reclass_line": False,
"template_gap": False,
"raw_value": None if raw_key is None else round(float(raw_key), 2),
# 对外调整额按呈现口径计算：期初未补（方向判定不可信）时不落地 → 0
"reclass_adjustment": None if raw_key is None else round(
float(plan["after"].get(key) or 0) - float(raw_key), 2),
float(_line_value(plan, key, "end") or 0) - float(raw_key), 2),
# 诊断用调整额（无论是否落地都给出，供底稿/审计追溯）
"diagnostic_reclass_adjustment": diag_adj,
})
# 规则落位项目（虚拟行）
@@ -1674,7 +1692,12 @@ def get_balance_sheet(
},
# ── 重分类（调表不调账）──
"reclass": {
"applied": bool(plan["entries"]),
# M2：期初未补 → 方向判定不可信 → 重分类结论只作底稿诊断，不落地到对外行项目
"applied": bool(plan["entries"]) and not diag_only,
"diagnostic_only": diag_only,
"direction_check_unreliable": diag_only,
"caliber_warning": plan.get("caliber_warning") or "",
"opening": plan.get("opening"),
"rule_version": plan["rule_version"],
"basis": plan["basis"],
"has_voucher_data": has_real,
@@ -1682,6 +1705,7 @@ def get_balance_sheet(
"invariants": inv,
"negatives": plan["negatives"],
"template_gaps": plan["template_gaps"],
"residual_reported": plan.get("residual_reported") or [],
"unmapped_subjects": plan["unmapped_subjects"],
"pl_unclosed": plan["pl_unclosed"],
"working_paper_url": f"/api/cma/reports/balance-sheet/reclass-working-paper?period={period}&basis={use_basis}",
@@ -1709,10 +1733,16 @@ def get_reclass_working_paper(
"rule_version": plan["rule_version"],
"generated_at": datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
"accounting_method": "调表不调账（报表取数层重分类，不生成记账凭证）",
# ── M1/M2：期初层 + 方向判定可信度（期初未补 → 只作诊断）──
"direction_check_unreliable": plan.get("direction_check_unreliable", False),
"caliber_warning": plan.get("caliber_warning") or "",
"opening": plan.get("opening"),
"diagnostic_only": bool(plan.get("direction_check_unreliable")),
"rules": plan["rules"],
"entries": plan["entries"],
"invariants": plan["invariants"],
"template_gaps": plan["template_gaps"],
"residual_reported": plan.get("residual_reported") or [],
"negatives": plan["negatives"],
"unmapped_subjects": plan["unmapped_subjects"],
"pl_unclosed": plan["pl_unclosed"],
@@ -1759,13 +1789,15 @@ def export_reclass_working_paper(
ws.column_dimensions[ws.cell(row=2, column=ci).column_letter].width = max(12, min(46, len(str(h)) * 2 + 10))
e_rows = report_reclass.entry_rows(plan)
warn = (" ⚠️ 期初未补，方向判定不可信（重分类结论仅诊断，不落地）"
if plan.get("direction_check_unreliable") else "")
ws1 = wb.active
ws1.title = "重分类底稿"
if e_rows:
_sheet(ws1, f"资产负债表重分类底稿（调表不调账） {period} 账套 entity_id={entity_id} 口径basis={use_basis} {plan['rule_version']}",
_sheet(ws1, f"资产负债表重分类底稿（调表不调账） {period} 账套 entity_id={entity_id} 口径basis={use_basis} {plan['rule_version']}{warn}",
list(e_rows[0].keys()), e_rows)
else:
ws1.cell(row=1, column=1, value=f"本期（{period}）无触发的重分类规则").font = Font(bold=True)
ws1.cell(row=1, column=1, value=f"本期（{period}）无触发的重分类规则{warn}").font = Font(bold=True)
d_rows = report_reclass.diagnostic_rows(plan)
ws2 = wb.create_sheet("不变量与诊断")
@@ -2091,6 +2123,10 @@ def export_statutory_reports(
"negatives": (bs.get("reclass") or {}).get("negatives") or {"before": [], "after": [], "after_in_scope": []},
"template_gaps": (bs.get("reclass") or {}).get("template_gaps") or [],
"unmapped_subjects": (bs.get("reclass") or {}).get("unmapped_subjects") or [],
# M1/M2：期初层 + 方向判定可信度（底稿 Sheet 同源暴露）
"residual_reported": (bs.get("reclass") or {}).get("residual_reported") or [],
"direction_check_unreliable": (bs.get("reclass") or {}).get("direction_check_unreliable", False),
"caliber_warning": (bs.get("reclass") or {}).get("caliber_warning", ""),
}
for row in report_reclass.diagnostic_rows(plan_like):
r = write_row(ws4, r, [row.get("项目"), row.get("值"), row.get("说明")])
backend/app/models/__init__.py
+1
View File
@@ -8,6 +8,7 @@ from app.models.knowledge import KnowledgeEvent, KnowledgeSummary
from app.models.driver_budget import DriverFactorTemplate, DriverFactorBudget
from app.models.product_sales import ProductSales
from app.models.ext_input_ledger import TaxFilingLog, HrMonthlyLedger, LicenceLedger
from app.models.report_opening import ReportOpeningBalance
from app.models.treasury import (
TreasuryAccount, TreasuryBalance, TreasuryControlRule,
TreasuryPayment, TreasuryControlHit, TreasuryLimit,
backend/app/models/report_opening.py
+41
View File
@@ -0,0 +1,41 @@
"""报表科目期初余额（账套建账期初层）— 2026-09-16
时点报表口径铁律（《CMA 资产负债表重分类规则表 v1.1》§零）：
余额 = 期初余额 + Σ(借−贷)（资产） / 期初余额 + Σ(贷−借)（负债·权益）
CMA 原有取数只有 `voucher_details` 累计发生额（账套最早期间 2021-03），**不含建账期初**，
使全部往来/税费类科目出现伪反向项（实测：1001 差 346,139.02、1405 差 6,133,427.20……逐月恒定）。
本表只作用于**报表取数层**：
· 不生成记账凭证、不写 `voucher_details`、不改 `kpi_values`；
· 期初数据来源：反解清单（status=derived，待源系统《科目余额表 2021.03》确认 → confirmed）。
"""
from sqlalchemy import Column, Integer, String, Numeric, DateTime, UniqueConstraint, func
from app.database import Base
class ReportOpeningBalance(Base):
"""账套建账期初余额（一账套一科目一期初行）"""
__tablename__ = "report_opening_balances"
__table_args__ = (
UniqueConstraint("entity_id", "subject_code", "effective_date",
name="uq_report_opening_entity_subject_effective"),
)
id = Column(Integer, primary_key=True, index=True)
entity_id = Column(Integer, nullable=False, index=True, comment="账套（实体）ID")
subject_code = Column(String(20), nullable=False, index=True, comment="科目编码（前缀匹配）")
subject_name = Column(String(100), comment="科目名称")
balance_side = Column(String(10), nullable=False, comment="余额方向 debit/credit")
amount = Column(Numeric(18, 2), nullable=False, default=0, comment="期初余额（正数）")
effective_date = Column(String(7), nullable=False, default="2021-03",
comment="生效期间 YYYY-MM（建账期）；早于该期的期间不套用")
source = Column(String(50), comment="来源标识，如 reverse-calc-20260916")
evidence = Column(String(255), comment="取证说明")
status = Column(String(20), nullable=False, default="derived",
comment="derived=反解待确认 / confirmed=源系统确认")
created_at = Column(DateTime, server_default=func.now())
updated_at = Column(DateTime, server_default=func.now(), onupdate=func.now())
backend/app/services/report_reclass.py
+178 -12
View File
@@ -1,14 +1,18 @@
"""资产负债表取数层重分类引擎（R1~R12，调表不调账）
依据：《CMA 资产负债表重分类规则表 v1》
（`wecom-finance/reports/CMA_资产负债表重分类规则表_v1.md`，任总 2026-09-16 批准落地）
依据：《CMA 资产负债表重分类规则表 v1.1》
（`wecom-finance/reports/CMA_资产负债表重分类规则表_v1.1.md`，2026-09-16；v1.0 已被替代）
本模块只作用于**报表取数层**：`voucher_details` 只读，**不生成记账凭证、不改账簿**
（调表不调账；账实不符走差错更正，见规则表 §四）。
口径要点
--------
1. **余额方向**：资产类 NSB = Σ(借−贷)；负债/权益类 NSB = Σ(贷−借)。NSB = normal-side balance。
0. **建账期初层（v1.1 §零 前置铁律）**：时点报表 余额 = 期初余额 + Σ(借−贷)（资产）/
期初余额 + Σ(贷−借)（负债·权益），期初取 `report_opening_balances`（只作用于报表取数，
不生成凭证、不写 voucher_details、不改 kpi_values）。期初缺失时**往来类方向判定不可信**
（`direction_check_unreliable=True`，报表层不得呈现"重分类转正"结论，只进底稿诊断）。
1. **余额方向**：资产类 NSB = 期初借 + Σ(借−贷)；负债/权益类 NSB = 期初贷 + Σ(贷−借)。
2. **触发**：NSB < 0（明细科目反方向余额，如预收账款出现借方余额）→ 转正额 = |NSB|，
移入目标报表项目（准则 30 号「不得抵销」）。
3. **全部触发判定基于原始余额**（单遍、顺序无关、可重复执行/幂等）——避免 R3/R4 这类互为镜像
@@ -32,7 +36,7 @@ from sqlalchemy import func
from app.utils.voucher_guard import apply_voucher_guard
RULE_VERSION = "reclass-v1.0-20260916"
RULE_VERSION = "reclass-v1.1-20260916"
DEFAULT_TOL = 0.01 # 元；小于该值的反方向余额不触发（浮点/分位噪声）
# 报表板块 → 余额方向
@@ -47,6 +51,66 @@ SIDE_OF_SECTION = {
# 资产备抵科目（余额在贷方）→ 取数方向反转：1602 累计折旧 / 1603 固定资产减值准备 / 1604 在建工程减值准备
CONTRA_CODES: Dict[str, str] = {"1602": "liability", "1603": "liability", "1604": "liability"}
# 账套建账期初层默认生效期间（早于该期的期间不套用期初；实际以表内 effective_date 为准）
DEFAULT_OPENING_EFFECTIVE_DATE = "2021-03"
def load_opening_rows(db, entity_id: int) -> List[dict]:
"""读取账套建账期初余额（`report_opening_balances`）。
期初层**只作用于报表取数**（余额 = 期初 + 累计发生额）：不生成凭证、不写 `voucher_details`。
表未建/无数据/查询异常 → 返回 []（此时 `direction_check_unreliable=True`，报表层不得呈现转正结论）。
"""
try:
from app.models import ReportOpeningBalance
rows = db.query(ReportOpeningBalance).filter(
ReportOpeningBalance.entity_id == entity_id
).all()
except Exception: # noqa: BLE001 - 表未建/异常 → 视为期初未导入
return []
out: List[dict] = []
for r in rows:
out.append({
"subject_code": str(r.subject_code or "").strip(),
"subject_name": r.subject_name or "",
"balance_side": str(r.balance_side or "").strip().lower(),
"amount": Decimal(str(r.amount or 0)),
"effective_date": str(r.effective_date or "").strip(),
"status": str(r.status or "").strip(),
"source": str(r.source or "").strip(),
})
return out
def _opening_summary(rows: List[dict]) -> dict:
"""期初层摘要（进底稿/接口，供财务核对借贷是否自平）。"""
debit = Decimal(0)
credit = Decimal(0)
dates, statuses, sources = set(), set(), set()
for r in rows:
if r["balance_side"] == "debit":
debit += r["amount"]
else:
credit += r["amount"]
if r["effective_date"]:
dates.add(r["effective_date"])
if r["status"]:
statuses.add(r["status"])
if r["source"]:
sources.add(r["source"])
return {
"imported": bool(rows),
"rows": len(rows),
"debit_total": float(round(debit, 2)),
"credit_total": float(round(credit, 2)),
"balanced": abs(debit - credit) < Decimal("0.01"),
"effective_dates": sorted(dates),
"statuses": sorted(statuses),
"sources": sorted(sources),
}
@dataclass(frozen=True)
class ReclassRule:
@@ -101,9 +165,11 @@ EXTRA_ITEMS: Dict[str, dict] = {
"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": [],
"OCA_OTHER": {"name": "其他流动资产", "section_key": "current_assets", "side": "asset",
"codes": [("1901", 1)],
"template_gap": True, "ns_category": "operating",
"note": "小企业准则项目池有「其他流动资产」，CMA 模板未列报（R9 落位项目）"},
"note": "小企业准则项目池有「其他流动资产」，CMA 模板未列报（R9 落位项目）；"
"并含 1901 待处理财产损溢的**过渡列报**（准则要求期末清理为零，待差错更正，M6）"},
"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": [],
@@ -118,9 +184,15 @@ EXTRA_ITEMS: Dict[str, dict] = {
# ── 余额取数（前缀匹配 + 作废守卫 + basis 口径） ─────────────────────────────
class BalanceSource:
"""按 (品类, 期间, basis) 取科目 normal-side balance（NSB），带会话内缓存。"""
"""按 (品类, 期间, basis) 取科目 normal-side balance（NSB），带会话内缓存。
def __init__(self, db, entity_id: int, period: str, basis: str = "end", tol: float = DEFAULT_TOL):
余额口径（v1.1 §零）= **期初余额（建账期初层）+ 累计发生额**：
· basis=end ：资产 NSB = 期初借 + Σ(借−贷)（累计至 period）；负债/权益 = 期初贷 + Σ(贷−借)
· basis=period ：只取本期发生额（**不含期初**，期初是时点余额概念）
"""
def __init__(self, db, entity_id: int, period: str, basis: str = "end", tol: float = DEFAULT_TOL,
opening_rows: Optional[List[dict]] = None):
self.db = db
self.entity_id = entity_id
self.period = period
@@ -128,6 +200,43 @@ class BalanceSource:
self.tol = tol
self._cache: Dict[Tuple[str, str], Optional[Decimal]] = {}
self._has_data: Optional[bool] = None
self._opening_rows: Optional[List[dict]] = opening_rows
# -- 建账期初层（M1） ---------------------------------------------------
def opening_rows(self) -> List[dict]:
if self._opening_rows is None:
self._opening_rows = load_opening_rows(self.db, self.entity_id)
return self._opening_rows
def has_opening(self) -> bool:
"""该账套是否已登记建账期初（未登记 → 方向判定不可信，M2）"""
return bool(self.opening_rows())
def opening_nsb(self, code: str, side: str, period: Optional[str] = None) -> Optional[Decimal]:
"""期初层对 NSB 的贡献；无该科目期初行 → None（与凭证层相加时按 0 处理）。
· 只对 basis=end 生效（时点余额口径）；
· 科目前缀匹配（期初行可挂在明细科目，如 `100101`）；
· `effective_date > period` 的行不套用（建账期之后的期间才用得上期初）。
"""
if self.basis != "end":
return None
period = period or self.period
total: Optional[Decimal] = None
for row in self.opening_rows():
sc = row["subject_code"]
if not sc or not sc.startswith(str(code)):
continue
eff = row["effective_date"]
if eff and period and eff > period:
continue
amt = row["amount"]
same_side = (
(row["balance_side"] == "debit" and side == "asset")
or (row["balance_side"] == "credit" and side in ("liability", "equity"))
)
total = (total if total is not None else Decimal(0)) + (amt if same_side else -amt)
return total
# -- 基础查询 ---------------------------------------------------------
def _query(self, code: str, period: str):
@@ -148,7 +257,7 @@ class BalanceSource:
return apply_voucher_guard(self.db, q, VoucherDetail)
def nsb(self, code: str, side: str, period: Optional[str] = None) -> Optional[Decimal]:
"""NSB = 资产 Σ(借−贷) / 负债·权益 Σ(贷−借)；无数据返回 None。"""
"""NSB = 期初 + 累计（资产 Σ(借−贷) / 负债·权益 Σ(贷−借)）；无凭证且无期初返回 None。"""
period = period or self.period
ck = (f"{code}|{side}|{period}|{self.basis}", "")
if ck in self._cache:
@@ -159,12 +268,16 @@ class BalanceSource:
d, c, n = row[0], row[1], row[2]
except Exception: # noqa: BLE001 - 无凭证明细表/查询异常 → 视为无数据
pass
opening = self.opening_nsb(code, side, period)
if not n:
self._cache[ck] = None
return None
# 无凭证明细：只有登记了期初才有余额，否则返回 None（回落示例口径）
self._cache[ck] = opening
return opening
debit = Decimal(str(d or 0))
credit = Decimal(str(c or 0))
val = (debit - credit) if side == "asset" else (credit - debit)
if opening is not None:
val += opening
self._cache[ck] = val
return val
@@ -261,6 +374,10 @@ def analyze(db, entity_id: int, period: str, basis: str = "end",
items = build_items(template_sections or [])
code_to_key = _covered_code_map(items)
# ── 建账期初层（M1）+ 方向判定可信度（M2） ──
opening_info = _opening_summary(bs.opening_rows())
direction_unreliable = not opening_info["imported"]
raw: Dict[str, Optional[Decimal]] = {}
prev: Dict[str, Optional[Decimal]] = {}
for key, item in items.items():
@@ -295,6 +412,7 @@ def analyze(db, entity_id: int, period: str, basis: str = "end",
"target_in_template": bool(tgt.get("template")),
"rule_status": r.status,
"error_check": r.error_check,
"diagnostic_only": direction_unreliable,
"note": r.note,
})
@@ -355,6 +473,13 @@ def analyze(db, entity_id: int, period: str, basis: str = "end",
"basis": bs.basis,
"period": period,
"rule_version": RULE_VERSION,
# ── 建账期初层 / 方向判定可信度（v1.1 §零，M1/M2）──
"direction_check_unreliable": direction_unreliable,
"opening_imported": opening_info["imported"],
"opening_rows": opening_info["rows"],
"opening_debit_total": opening_info["debit_total"],
"opening_credit_total": opening_info["credit_total"],
"opening_balanced": opening_info["balanced"],
"assets_before": before_t["assets"],
"assets_after": after_t["assets"],
"liabilities_before": before_t["liabilities"],
@@ -426,6 +551,15 @@ def analyze(db, entity_id: int, period: str, basis: str = "end",
"entity_id": entity_id,
"prev_period": _prev_period(period),
"has_voucher_data": bs.has_data(),
# ── 期初层与可信度（M1/M2）──
"direction_check_unreliable": direction_unreliable,
"caliber_warning": (
"期初未补，方向判定不可信：本账套无建账期初余额（report_opening_balances），"
"余额口径缺少期初层 → 负数项目/反方向项判定不得作为对外列报依据；"
"重分类触发结论仅供底稿诊断，不进对外行项目（规则表 v1.1 §零-4）。"
if direction_unreliable else ""
),
"opening": opening_info,
"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()},
@@ -437,6 +571,15 @@ def analyze(db, entity_id: int, period: str, basis: str = "end",
"totals_after": after_t,
"negatives": {"before": neg_before, "after": neg_after, "after_in_scope": neg_after_in_scope},
"template_gaps": template_gaps,
# 规则池未列报、但按准则项目池**过渡列报**的科目（M6：1901 待处理财产损溢 → 其他流动资产）
"residual_reported": [
{"item_key": key, "name": item["name"],
"subject_codes": [c for c, s in item["codes"] if s == 1],
"value": (float(raw[key]) if raw.get(key) is not None else None),
"note": item.get("note", "")}
for key, item in items.items()
if "1901" in [c for c, s in item["codes"] if s == 1]
],
"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,
}
@@ -559,6 +702,7 @@ def entry_rows(analysis: dict) -> List[dict]:
"目标科目": "/".join(e["target_codes"]) or "-",
"目标项模板已列报": "是" if e["target_in_template"] else "否(虚拟行)",
"处理": "调表不调账",
"诊断模式": "是（期初未补，方向判定不可信→不落地）" if e.get("diagnostic_only") else "否",
"需先行差错核查": "是" if e["error_check"] else "否",
"依据/备注": e["note"],
})
@@ -570,6 +714,17 @@ def diagnostic_rows(analysis: dict) -> List[dict]:
inv = analysis["invariants"]
rows = [
{"项目": "规则版本", "值": inv["rule_version"], "说明": "取数层重分类规则集"},
{"项目": "建账期初层（时点报表口径）",
"值": (f"已导入 {inv.get('opening_rows', 0)} 行"
if inv.get("opening_imported") else "未导入"),
"说明": ("余额 = 期初 + 累计发生额；期初只作用于报表取数，不生成凭证、不写 voucher_details。"
f"期初借贷合计 {inv.get('opening_debit_total', 0)} / {inv.get('opening_credit_total', 0)}"
f"（自平={'是' if inv.get('opening_balanced') else '否'}）"
if inv.get("opening_imported") else
"本账套无建账期初余额 → 余额口径缺期初层（v1.1 §零）")},
{"项目": "方向判定可信度",
"值": "不可信（期初未补）" if analysis.get("direction_check_unreliable") else "可信",
"说明": analysis.get("caliber_warning") or "期初已补，方向判定以「期初 + 累计」余额为准"},
{"项目": "取数口径 basis", "值": inv["basis"], "说明": "end=期末余额(累计至期间)；period=本期发生额"},
{"项目": "触发规则数", "值": inv["entries_count"], "说明": "本轮实际触发的重分类条数"},
{"项目": "转正额合计(元)", "值": inv["reclass_total"], "说明": "调表不调账，不生成凭证"},
@@ -589,7 +744,8 @@ def diagnostic_rows(analysis: dict) -> List[dict]:
{"项目": "总资产不变", "值": "是" if inv["total_assets_unchanged"] else "否（跨方向重分类必然变化）",
"说明": inv["total_assets_unchanged_semantics"]},
{"项目": "规则范围内负数项目(重分类后)", "值": len(analysis["negatives"]["after_in_scope"]),
"说明": "应为 0"},
"说明": ("应为 0" if not analysis.get("direction_check_unreliable")
else "诊断用（期初未补，方向判定不可信，不得作为对外列报依据）")},
{"项目": "负数项目总数(重分类前/后)", "值": f"{inv['negatives_before']} / {inv['negatives_after']}",
"说明": "重分类范围外科目见下方清单"},
]
@@ -612,6 +768,16 @@ def diagnostic_rows(analysis: dict) -> List[dict]:
if abs(plu) > 0.01 and abs(plu - inv.get("residual_after", 0)) < 0.01
else ("非 0，需差错核查" if abs(plu) > 0.01 else "已结平")),
})
# 1901 待处理财产损溢 → 过渡列报「其他流动资产」以保恒等式（v1.1 §三 补注，M6）
for rl in (analysis.get("residual_reported") or []):
if not rl.get("value"):
continue
rows.append({
"项目": f"过渡列报：{'/'.join(rl.get('subject_codes') or [])} 待处理财产损溢",
"值": round(float(rl["value"]), 2),
"说明": "准则项目池无「待处理财产损溢」正式行 → 过渡期按「其他流动资产」列报以保恒等式；"
"准则要求该科目期末清理为零，未清理余额即勾稽差额来源，属财务差错更正事项（M6）",
})
for u in analysis["unmapped_subjects"]:
rows.append({
"项目": f"未列报科目：{u['subject_code']}",
backend/scripts/import_report_opening_balances.py
+150
View File
@@ -0,0 +1,150 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""账套建账期初余额导入器（M1）—— 幂等 / 默认 dry-run / 显式 --commit 才写库
来源清单：`wecom-finance/reports/CMA_酣客建账期初余额_反解清单_v1.csv`（13 行，借贷各 15,110,544.36）
写入表：`report_opening_balances`（只作用于报表取数；**不生成凭证、不写 voucher_details、不改 kpi_values**）
用法：
cd /root/cma-management/backend
venv/bin/python3 scripts/import_report_opening_balances.py # dry-run（默认，零写入）
venv/bin/python3 scripts/import_report_opening_balances.py --commit # 落库（需放行）
venv/bin/python3 scripts/import_report_opening_balances.py --commit --entity-id 1 --effective-date 2021-03
幂等：按 (entity_id, subject_code, effective_date) 判存 —— 新增/等值跳过/差异更新，重复执行结果一致。
放行前置：本仓 push 会触发部署重启（生产变更），须任总放行；表由 `init_db()`（应用启动 create_all）建出。
"""
from __future__ import annotations
import argparse
import csv
import io
import os
import sys
from decimal import Decimal
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
DEFAULT_CSV = "/root/.hermes/profiles/wecom-finance/reports/CMA_酣客建账期初余额_反解清单_v1.csv"
DEFAULT_SOURCE = "reverse-calc-20260916"
DEFAULT_STATUS = "derived" # 待源系统《科目余额表 2021.03》确认后升级 confirmed
def read_csv(path: str):
rows = []
with io.open(path, encoding="utf-8-sig") as f:
for r in csv.DictReader(f):
rows.append({
"entity_id": int((r.get("entity_id") or "1").strip()),
"subject_code": (r.get("subject_code") or "").strip(),
"subject_name": (r.get("subject_name") or "").strip(),
"balance_side": (r.get("balance_side") or "").strip().lower(),
"amount": Decimal(str((r.get("opening_amount") or "0").strip() or 0)),
"evidence": (r.get("evidence") or "").strip()[:255],
"status_note": (r.get("status") or "").strip(),
})
return rows
def main():
ap = argparse.ArgumentParser()
ap.add_argument("--csv", default=DEFAULT_CSV)
ap.add_argument("--entity-id", type=int, default=None, help="覆盖 CSV 内的 entity_id")
ap.add_argument("--effective-date", default="2021-03", help="建账期 YYYY-MM（期初生效期间）")
ap.add_argument("--source", default=DEFAULT_SOURCE)
ap.add_argument("--status", default=DEFAULT_STATUS, choices=["derived", "confirmed"])
ap.add_argument("--commit", action="store_true", help="真正写库（缺省=dry-run，零写入）")
args = ap.parse_args()
rows = read_csv(args.csv)
if args.entity_id is not None:
for r in rows:
r["entity_id"] = args.entity_id
debit = sum((r["amount"] for r in rows if r["balance_side"] == "debit"), Decimal(0))
credit = sum((r["amount"] for r in rows if r["balance_side"] == "credit"), Decimal(0))
print(f"[清单] {args.csv}")
print(f"[清单] {len(rows)} 行 ｜ 借 {debit:,.2f} / 贷 {credit:,.2f} / 差 {debit - credit:,.2f} "
f"（自平={'是' if abs(debit - credit) < Decimal('0.01') else '否'}）")
for r in rows:
print(f" e{r['entity_id']} {r['subject_code']:<6}{r['subject_name']:<10}"
f"{r['balance_side']:<7}{r['amount']:>16,.2f} {r['status_note']}")
if abs(debit - credit) >= Decimal("0.01"):
print("[FAIL] 期初借贷不平衡 —— 拒绝导入")
return 2
from app.database import get_session_local
try:
from app.models import ReportOpeningBalance
except Exception as exc: # noqa: BLE001
print(f"[FAIL] 无法加载模型（表未建？先应用启动 create_all）：{exc}")
return 3
db = get_session_local()()
try:
# 预检：表是否存在/现有行数（dry-run 也读，只读不写）
try:
existing = db.query(ReportOpeningBalance).all()
table_ready = True
except Exception as exc: # noqa: BLE001 - 表未建（部署前 create_all 才会建）
existing = []
table_ready = False
print(f"[预检] report_opening_balances 不可读（{type(exc).__name__}）→ 视为 0 行；"
f"表由部署后应用启动 `init_db()`/create_all 建出")
idx = {(r.entity_id, r.subject_code, r.effective_date): r for r in existing}
print(f"[预检] report_opening_balances 现有 {len(existing)} 行（表就绪={table_ready}）；"
f"将按 (entity_id, subject_code, effective_date={args.effective_date}) 幂等导入")
ins = upd = skip = 0
plan = []
for r in rows:
key = (r["entity_id"], r["subject_code"], args.effective_date)
cur = idx.get(key)
if cur is None:
ins += 1
plan.append(("INSERT", r))
elif (str(cur.balance_side) == r["balance_side"]
and Decimal(str(cur.amount)) == r["amount"]
and str(cur.status) == args.status):
skip += 1
plan.append(("SKIP", r))
else:
upd += 1
plan.append(("UPDATE", r))
if args.commit:
if cur is None:
db.add(ReportOpeningBalance(
entity_id=r["entity_id"], subject_code=r["subject_code"],
subject_name=r["subject_name"], balance_side=r["balance_side"],
amount=r["amount"], effective_date=args.effective_date,
source=args.source, evidence=r["evidence"] or args.source,
status=args.status,
))
else:
cur.subject_name = r["subject_name"]
cur.balance_side = r["balance_side"]
cur.amount = r["amount"]
cur.source = args.source
cur.evidence = r["evidence"] or args.source
cur.status = args.status
print(f"[计划] 新增 {ins} / 更新 {upd} / 跳过（等值）{skip}"
f" ｜ 模式={'COMMIT（写库）' if args.commit else 'DRY-RUN（零写入）'}")
if not args.commit:
print("[dry-run] 未写库；确认无误后加 --commit 执行（须任总放行后）")
return 0
db.commit()
after = db.query(ReportOpeningBalance).filter(
ReportOpeningBalance.entity_id.in_({r["entity_id"] for r in rows})
).all()
d2 = sum((Decimal(str(x.amount)) for x in after if x.balance_side == "debit"), Decimal(0))
c2 = sum((Decimal(str(x.amount)) for x in after if x.balance_side == "credit"), Decimal(0))
print(f"[回查] 落库后 {len(after)} 行 ｜ 借 {d2:,.2f} / 贷 {c2:,.2f} / 差 {d2 - c2:,.2f}")
ok = len(after) == len(rows) and abs(d2 - debit) < Decimal("0.01") and abs(c2 - credit) < Decimal("0.01")
print("[PASS] 落库回查一致" if ok else "[FAIL] 落库回查不一致")
return 0 if ok else 4
finally:
db.close()
if __name__ == "__main__":
sys.exit(main())
backend/scripts/verify_opening_balance_live_20260916.py
+262
View File
@@ -0,0 +1,262 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""建账期初层（M1/M2）现网只读验收 —— 2026-09-16
判据（任务书 `finance-20260916-opening-balance-layer.md` 验收标准）：
① 13 科目期末余额 = 期初 + 累计，与源系统《科目余额表》2026.08 / 2026.07 **逐科目 0 差异**（容差 ≤0.01）
② 恒等式：资产 = 负债 + 权益，勾稽差额 0.00（含 1901 按「其他流动资产」列报）
③ M7：权益合计 10,932,589.50（含实收资本 3001 = 2,000,000.00）
④ 期初补齐后 R1~R12 触发数 = 0（R2/R4 伪触发消失）
⑤ M2 两态：期初未补 → direction_check_unreliable=True；期初注入 → 自动 False
**零写入**：全程只读 SELECT；期初行不落库（用 monkeypatch 注入内存期初，不建表、不 INSERT、不 commit）。
用法：cd /root/cma-management/backend && venv/bin/python3 scripts/verify_opening_balance_live_20260916.py [period]
"""
from __future__ import annotations
import csv
import io
import json
import os
import sys
from decimal import Decimal
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
from sqlalchemy import func # noqa: E402
from app.database import get_session_local # noqa: E402
from app.models import VoucherDetail # noqa: E402
from app.services import report_reclass as rr # noqa: E402
from app.utils.voucher_guard import apply_voucher_guard # noqa: E402
PERIOD = sys.argv[1] if len(sys.argv) > 1 else "2026-08"
PREV = "2026-07"
ENTITY = 1
SRC = "/root/WB_WS/_data/zxt-hanke/%s/kmyeb.csv"
OPENING_CSV = "/root/.hermes/profiles/wecom-finance/reports/CMA_酣客建账期初余额_反解清单_v1.csv"
OUT_JSON = "/root/.hermes/profiles/wecom-fullstack/pending-tasks/_reports/finance-20260916-opening-balance-verify.json"
TOL = Decimal("0.01")
# 任务书给出的 13 科目期末余额（借−贷，负=贷方）
EXPECT = {
"1001": Decimal("6331.30"), "1002": Decimal("46729.79"), "1122": Decimal("1449489.54"),
"1123": Decimal("2066724.00"), "1405": Decimal("5462698.72"), "1601": Decimal("9333477.39"),
"1602": Decimal("-516957.04"), "1801": Decimal("0.00"), "2203": Decimal("-836505.86"),
"2221": Decimal("-4295885.13"), "3001": Decimal("-2000000.00"), "3103": Decimal("-14918101.66"),
"3104": Decimal("5985512.16"),
}
SRC_TOTALS = {"assets": Decimal("18203980.49"), "liabilities": Decimal("7271390.99"),
"equity": Decimal("10932589.50")}
FINDINGS = []
def chk(status: str, name: str, detail: str):
FINDINGS.append({"status": status, "judge": name, "detail": detail})
print(f"[{status}] {name} — {detail}")
def load_opening():
rows = []
with io.open(OPENING_CSV, encoding="utf-8-sig") as f:
for r in csv.DictReader(f):
rows.append({
"subject_code": (r["subject_code"] or "").strip(),
"subject_name": (r["subject_name"] or "").strip(),
"balance_side": (r["balance_side"] or "").strip().lower(),
"amount": Decimal(str(r["opening_amount"]).strip()),
"effective_date": "2021-03",
"status": "derived", "source": "reverse-calc-20260916",
})
return rows
OPENING = load_opening()
def src_rows(period: str):
"""源系统《科目余额表》4 位科目行 → {code: (name, 期末借−期末贷)}"""
path = SRC % period.replace("-", ".")
out = {}
if not os.path.exists(path):
return out
with io.open(path, encoding="utf-8-sig") as f:
for r in csv.DictReader(f):
code = (r["科目编码"] or "").strip()
if len(code) != 4:
continue
d = Decimal(str(r["期末借"] or "0"))
c = Decimal(str(r["期末贷"] or "0"))
out[code] = ((r["科目名称"] or "").strip(), d - c)
return out
def src_totals(rows):
t = {"assets": Decimal(0), "liabilities": Decimal(0), "equity": Decimal(0)}
for code, (_n, net) in rows.items():
if code.startswith("1"):
t["assets"] += net
elif code.startswith("2"):
t["liabilities"] += -net
elif code.startswith("3"):
t["equity"] += -net
return t
def cma_cum(db, code: str, period: str) -> Decimal:
"""CMA voucher_details 累计（借−贷，作废守卫生效）"""
q = db.query(func.sum(VoucherDetail.debit_amount - VoucherDetail.credit_amount)).filter(
VoucherDetail.entity_id == ENTITY,
VoucherDetail.subject_code.like(f"{code}%"),
VoucherDetail.period <= period,
)
v = apply_voucher_guard(db, q, VoucherDetail).scalar()
return Decimal(str(v or 0))
def opening_signed(code: str) -> Decimal:
tot = Decimal(0)
for r in OPENING:
if r["subject_code"].startswith(code):
tot += r["amount"] if r["balance_side"] == "debit" else -r["amount"]
return tot
def main():
db = get_session_local()()
evidence = {"period": PERIOD, "prev_period": PREV, "entity_id": ENTITY, "mode": "read-only",
"opening_rows": len(OPENING), "subjects": [], "checks": []}
try:
# ── ① 13 科目：期初 + 累计 vs 源系统期末（2026.08 / 2026.07 双期） ──
for period in (PERIOD, PREV):
sr = src_rows(period)
bad = 0
print(f"\n=== [①] {period} 13 科目 期初 + 累计 vs 源系统《科目余额表》期末（借−贷）===")
print(f"{'科目':<6}{'名称':<12}{'期初':>16}{'CMA累计':>16}{'期初+累计':>16}"
f"{'源系统期末':>16}{'差异':>12} 判定")
for code in list(EXPECT) if period == PERIOD else list(EXPECT):
name, snet = sr.get(code, ("", None))
cum = cma_cum(db, code, period)
op = opening_signed(code)
total = cum + op
diff = (total - snet) if snet is not None else None
ok = diff is not None and abs(diff) <= TOL
if period == PERIOD:
evidence["subjects"].append({
"subject_code": code, "subject_name": name, "period": period,
"opening": float(op), "cma_cumulative": float(cum),
"opening_plus_cumulative": float(total),
"source_end_balance": float(snet) if snet is not None else None,
"diff": float(diff) if diff is not None else None, "pass": bool(ok),
})
if not ok:
bad += 1
print(f"{code:<6}{name:<12}{float(op):>16,.2f}{float(cum):>16,.2f}{float(total):>16,.2f}"
f"{(float(snet) if snet is not None else float('nan')):>16,.2f}"
f"{(float(diff) if diff is not None else float('nan')):>12,.2f} {'OK' if ok else 'FAIL'}")
chk("PASS" if bad == 0 else "FAIL",
f"{period} 13 科目逐科目 0 差异",
f"不一致 {bad} 项（容差 {TOL}）")
# ── 任务书硬判据：13 科目期末余额逐科目对表 ──
bad_exp = 0
for code, want in EXPECT.items():
cum = cma_cum(db, code, PERIOD) + opening_signed(code)
if abs(cum - want) > TOL:
bad_exp += 1
print(f" [差异] {code}: {cum} != {want}")
chk("PASS" if bad_exp == 0 else "FAIL", "任务书 13 科目期末余额硬判据",
f"不一致 {bad_exp} 项")
# ── 源系统合计（4 位科目自算，独立于财务给出值） ──
st = src_totals(src_rows(PERIOD))
ok_src = all(abs(st[k] - SRC_TOTALS[k]) <= TOL for k in SRC_TOTALS)
chk("PASS" if ok_src else "FAIL", "源系统合计复算可比",
f"资产 {st['assets']:,.2f} / 负债 {st['liabilities']:,.2f} / 权益 {st['equity']:,.2f}")
evidence["source_totals_selfcalc"] = {k: float(v) for k, v in st.items()}
# ── ②③④ 报表层：期初注入（内存，不落库）后跑引擎 ──
from app.api.reports import BALANCE_SHEET_SECTIONS
def analyze_with(opening_rows):
orig = rr.load_opening_rows
rr.load_opening_rows = lambda db_, entity_id: list(opening_rows)
try:
return rr.analyze(db, ENTITY, PERIOD, basis="end",
template_sections=BALANCE_SHEET_SECTIONS)
finally:
rr.load_opening_rows = orig
p_no = analyze_with([]) # 态①：期初未补（现网实况）
p_yes = analyze_with(OPENING) # 态②：期初补齐（F1 反解值）
evidence["state_no_opening"] = {
"direction_check_unreliable": p_no["direction_check_unreliable"],
"entries_count": p_no["invariants"]["entries_count"],
"entries": [(e["rule"], e["reclass_amount"], e["source_code"]) for e in p_no["entries"]],
"totals_after": p_no["totals_after"],
}
evidence["state_with_opening"] = {
"direction_check_unreliable": p_yes["direction_check_unreliable"],
"opening": p_yes["opening"],
"entries_count": p_yes["invariants"]["entries_count"],
"entries": [(e["rule"], e["reclass_amount"], e["source_code"]) for e in p_yes["entries"]],
"totals_after": p_yes["totals_after"],
"invariants": {k: p_yes["invariants"][k] for k in (
"assets_after", "liabilities_after", "equity_after", "residual_after",
"identity_holds_after", "opening_imported", "opening_balanced")},
}
print(f"\n=== [②③④] 报表层（basis=end, {PERIOD}）===")
print(f" 期初未补：可信度不可信={p_no['direction_check_unreliable']} 触发={p_no['invariants']['entries_count']} "
f"资产={p_no['totals_after']['assets']:,.2f} 负债={p_no['totals_after']['liabilities']:,.2f} "
f"权益={p_no['totals_after']['equity']:,.2f} 差额={p_no['totals_after']['residual']:,.2f}")
ta = p_yes["totals_after"]
print(f" 期初补齐：可信度不可信={p_yes['direction_check_unreliable']} 触发={p_yes['invariants']['entries_count']} "
f"资产={ta['assets']:,.2f} 负债={ta['liabilities']:,.2f} 权益={ta['equity']:,.2f} "
f"差额={ta['residual']:,.2f}")
print(f" 期初层：{p_yes['opening']}")
ok_tot = (abs(Decimal(str(ta["assets"])) - SRC_TOTALS["assets"]) <= TOL
and abs(Decimal(str(ta["liabilities"])) - SRC_TOTALS["liabilities"]) <= TOL
and abs(Decimal(str(ta["equity"])) - SRC_TOTALS["equity"]) <= TOL)
chk("PASS" if ok_tot else "FAIL", "② 三合计与源系统一致",
f"资产 {ta['assets']:,.2f}(源 18,203,980.49) / 负债 {ta['liabilities']:,.2f}(源 7,271,390.99) "
f"/ 权益 {ta['equity']:,.2f}(源 10,932,589.50)")
chk("PASS" if abs(Decimal(str(ta["residual"]))) <= TOL else "FAIL", "② 勾稽差额 0.00",
f"residual_after={ta['residual']}")
chk("PASS" if abs(Decimal(str(ta["equity"])) - SRC_TOTALS["equity"]) <= TOL else "FAIL",
"③ M7 权益合计（含 3001 实收资本 2,000,000）",
f"权益合计 {ta['equity']:,.2f}（实收资本行 raw={p_yes['raw'].get('4001')}）")
chk("PASS" if p_yes["invariants"]["entries_count"] == 0 else "FAIL", "④ R1~R12 触发数 = 0（R2/R4 伪触发消失）",
f"期初补齐后触发 {p_yes['invariants']['entries_count']} 条 / 期初未补时 {p_no['invariants']['entries_count']} 条（诊断）")
chk("PASS" if (p_no["direction_check_unreliable"] and not p_yes["direction_check_unreliable"]) else "FAIL",
"⑤ M2 两态（期初未补=不可信 → 补齐=可信）",
f"态①{ p_no['direction_check_unreliable'] } → 态②{ p_yes['direction_check_unreliable'] }")
# ── 零持久化自证 ──
try:
from app.models import ReportOpeningBalance
n_rows = db.query(ReportOpeningBalance).count()
has_table = True
except Exception as exc: # noqa: BLE001
n_rows, has_table = None, False
print(f" [INFO] report_opening_balances 未建（{type(exc).__name__}）")
chk("PASS", "零持久化（期初未写入现网库）",
f"本次仅内存注入；表存在={has_table} 行数={n_rows}（未落库）")
evidence["checks"] = FINDINGS
evidence["pass"] = all(f["status"] == "PASS" for f in FINDINGS)
os.makedirs(os.path.dirname(OUT_JSON), exist_ok=True)
with io.open(OUT_JSON, "w", encoding="utf-8") as f:
json.dump(evidence, f, ensure_ascii=False, indent=2)
print(f"\n证据 JSON: {OUT_JSON} ({os.path.getsize(OUT_JSON)} B)")
n_fail = sum(1 for f in FINDINGS if f["status"] == "FAIL")
print(f"总判定：PASS {len(FINDINGS) - n_fail} / FAIL {n_fail}")
return 1 if n_fail else 0
finally:
db.close()
if __name__ == "__main__":
sys.exit(main())
backend/tests/test_report_reclass.py
+155 -7
View File
@@ -84,6 +84,32 @@ def _analyze(db, period=PERIOD, entity_id=1, basis="end"):
return rr.analyze(db, entity_id, period, basis=basis, template_sections=BALANCE_SHEET_SECTIONS)
# ── 建账期初层夹具（M1，2026-09-16 v1.1） ────────────────────────────────────
def seed_opening(db: Session, entity_id: int = 1, rows=None, effective_date: str = "2021-03"):
"""登记账套建账期初（`report_opening_balances`）。
期初层只作用于报表取数（余额 = 期初 + 累计），不生成凭证、不写 voucher_details。
rows: [(subject_code, subject_name, balance_side, amount)]
"""
from app.models import ReportOpeningBalance
for code, name, side, amt in (rows or []):
db.add(ReportOpeningBalance(
entity_id=entity_id, subject_code=code, subject_name=name,
balance_side=side, amount=amt, effective_date=effective_date,
source="test-fixture", evidence="unit-test", status="derived",
))
db.commit()
def mark_opening_imported(db: Session, entity_id: int = 1):
"""登记"期初已补（显式为零）"标记行 —— M2 可信度两态测试用。
零额登记行也是"该账套期初已核对"的合法表达（表内有行 ⇒ direction_check_unreliable=False）。
"""
seed_opening(db, entity_id, [("1001", "库存现金", "debit", 0)])
class TestReclassEngine:
def test_t1_triggers_and_reversal(self, db: Session):
seed_reverse_items(db)
@@ -221,6 +247,7 @@ class TestReclassEndpoints:
def test_t7b_reclass_applied_on_real_data(self, client: TestClient, db: Session):
seed_reverse_items(db)
mark_opening_imported(db) # v1.1：期初已登记（方向判定可信）才呈现转正
h = self._auth(client, db)
data = client.get(f"{BASE}/balance-sheet?period={PERIOD}&basis=end", headers=h).json()
assert data["basis"] == "end"
@@ -241,6 +268,7 @@ class TestReclassEndpoints:
def test_t2b_virtual_line_rendered_in_section(self, client: TestClient, db: Session):
seed_receivable_side(db)
mark_opening_imported(db) # v1.1：期初登记后虚拟行才作为对外列报呈现
create_test_user(db)
h = auth_header(get_token_for_user(client))
data = client.get(f"{BASE}/balance-sheet?period={PREV}", headers=h).json()
@@ -282,15 +310,135 @@ class TestReclassEndpoints:
class TestUnmappedDiagnosis:
def test_unmapped_subjects_and_gap_explained(self, db: Session):
"""模板未列报科目 → 出现在诊断里，且勾稽差额与之一致"""
"""1901 待处理财产损溢：v1.1 起过渡列报「其他流动资产」→ 不再是未列报科目，恒等式闭合"""
_v(db, 1, PERIOD, [("1002", 1_000, 0), ("4001", 0, 1_000)])
_v(db, 1, PERIOD, [("1801", 200, 0), ("1002", 0, 200)]) # 长期待摊费用（新补的行，不再未列报）
_v(db, 1, PERIOD, [("1901", 300, 0), ("1002", 0, 300)]) # 待处理财产损溢（模板未列报）
_v(db, 1, PERIOD, [("1801", 200, 0), ("1002", 0, 200)]) # 长期待摊费用（模板已含）
_v(db, 1, PERIOD, [("1901", 300, 0), ("1002", 0, 300)]) # 待处理财产损溢（v1.1：过渡列报）
db.add(Subject(subject_code="1901", subject_name="待处理财产损溢", category="asset"))
db.add(Subject(subject_code="190102", subject_name="待处理流动资产损溢", category="detail"))
db.commit()
mark_opening_imported(db)
p = _analyze(db)
codes = [u["subject_code"] for u in p["unmapped_subjects"]]
assert "1901" in codes
unmapped_sum = sum(u["normal_side_balance"] for u in p["unmapped_subjects"])
assert p["invariants"]["residual_after"] == pytest.approx(-unmapped_sum)
assert "1901" not in codes # v1.1：已按「其他流动资产」列报
assert p["raw"]["OCA_OTHER"] == pytest.approx(300.0)
# 过渡列报后恒等式闭合（勾稽差额 0），且底稿如实标注"待清理"
assert p["invariants"]["residual_after"] == pytest.approx(0.0)
assert p["invariants"]["identity_holds_after"] is True
diag = {r["项目"]: r for r in rr.diagnostic_rows(p)}
assert "过渡列报：1901 待处理财产损溢" in diag
assert "清理为零" in diag["过渡列报：1901 待处理财产损溢"]["说明"]
# ── M1/M2/M7（《CMA 资产负债表重分类规则表 v1.1》§零/§三，2026-09-16） ────────
class TestOpeningBalanceLayer:
"""建账期初层（余额 = 期初 + 累计）+ 方向判定可信度 + 权益行科目集"""
def test_t11_opening_plus_cumulative(self, db: Session):
"""M1：余额 = 期初 + 累计；期初只作用于 basis=end，且受 effective_date 约束"""
_v(db, 1, PREV, [("1002", 400_000, 0), ("2203", 0, 400_000)])
_v(db, 1, PERIOD, [("2203", 500_000, 0), ("1002", 0, 500_000)])
seed_opening(db, 1, [
("2203", "预收账款", "credit", 1_000_000),
("1601", "固定资产", "debit", 286_847.83),
("1602", "累计折旧", "credit", 84_690.11),
("1002", "银行存款", "debit", 797_842.28),
])
p = _analyze(db, period=PERIOD, basis="end")
# 2203 = 期初贷 1,000,000 + (贷 400,000 − 借 500,000) = 900,000（贷方，正常）→ 伪触发 R2 消失
assert p["raw"]["2203"] == pytest.approx(900_000.0)
assert p["entries"] == []
assert p["invariants"]["opening_imported"] is True
assert p["invariants"]["opening_rows"] == 4
assert p["invariants"]["opening_balanced"] is True
# 备抵科目（1602 累计折旧，贷方）反向取数：固定资产净额 = 原值 − 累计折旧
assert p["raw"]["1601|1602"] == pytest.approx(286_847.83 - 84_690.11)
# basis=period（本期发生额）不含期初 —— 期初是时点余额概念
per = _analyze(db, period=PERIOD, basis="period")
assert per["raw"]["2203"] == pytest.approx(-500_000.0)
assert per["invariants"]["opening_imported"] is True # 期初仍在（只是不被套用）
# effective_date 早于建账期 → 不套用期初
bs = rr.BalanceSource(db, 1, "2021-02", "end")
assert bs.opening_nsb("2203", "liability") is None
bs2 = rr.BalanceSource(db, 1, "2021-03", "end")
assert bs2.opening_nsb("2203", "liability") == pytest.approx(1_000_000.0)
def test_t12_direction_unreliable_two_states(self, client: TestClient, db: Session):
"""M2：期初未补 → direction_check_unreliable=True 且不呈现"重分类转正"；期初补 → 自动转 False"""
seed_reverse_items(db)
create_test_user(db)
h = auth_header(get_token_for_user(client))
# 态①：期初未补（现有实体实况）
p = _analyze(db)
assert p["direction_check_unreliable"] is True
assert p["caliber_warning"].startswith("期初未补，方向判定不可信")
assert all(e["diagnostic_only"] for e in p["entries"])
d = client.get(f"{BASE}/balance-sheet?period={PERIOD}&basis=end", headers=h).json()
assert d["reclass"]["applied"] is False
assert d["reclass"]["direction_check_unreliable"] is True
assert d["reclass"]["opening"]["imported"] is False
lines = {ln["item_key"]: ln for sec in d["sections"] for ln in sec["lines"]}
# 对外行项目 = 原始余额（不转正）；触发只进底稿诊断
assert lines["2203"]["end_value"] == pytest.approx(-100_000.0)
assert lines["2203"]["reclass_adjustment"] == pytest.approx(0.0)
assert lines["2203"]["diagnostic_reclass_adjustment"] == pytest.approx(100_000.0)
wp = client.get(f"{BASE}/balance-sheet/reclass-working-paper?period={PERIOD}", headers=h).json()
assert wp["direction_check_unreliable"] is True and wp["diagnostic_only"] is True
assert len(wp["entries"]) == 7 # 诊断条目仍在底稿
assert all(e["diagnostic_only"] for e in wp["entries"])
# 态②：期初补齐（2203 期初贷 400,000 + 累计 −100,000 = 300,000 正常）→ 伪触发消失、可信
seed_opening(db, 1, [("2203", "预收账款", "credit", 400_000)])
p2 = _analyze(db)
assert p2["direction_check_unreliable"] is False
assert p2["caliber_warning"] == ""
d2 = client.get(f"{BASE}/balance-sheet?period={PERIOD}&basis=end", headers=h).json()
assert d2["reclass"]["direction_check_unreliable"] is False
assert d2["reclass"]["opening"]["imported"] is True
lines2 = {ln["item_key"]: ln for sec in d2["sections"] for ln in sec["lines"]}
assert lines2["2203"]["end_value"] == pytest.approx(300_000.0)
assert "2203" not in {e["rule"] for e in d2["reclass"]["entries"]}
def test_t13_idempotent_and_readonly(self, db: Session):
"""幂等（同参数重复分析一致）+ 只读（期初层不生成凭证、不写 voucher_details）"""
seed_reverse_items(db)
seed_opening(db, 1, [("2203", "预收账款", "credit", 400_000)])
n_voucher = db.query(VoucherDetail).count()
a = _analyze(db)
b = _analyze(db)
assert a["raw"] == b["raw"] and a["after"] == b["after"]
assert [e["rule"] for e in a["entries"]] == [e["rule"] for e in b["entries"]]
assert db.query(VoucherDetail).count() == n_voucher # 凭证零写入
from app.models import ReportOpeningBalance
assert db.query(ReportOpeningBalance).count() == 1 # 期初行未被分析改写
def test_t14_equity_code_set(self, db: Session):
"""M7：权益行科目集 = {3001,3002,3101,3103,3104,3105}（含 4001/4002/4101/4103/4104 别名）"""
_v(db, 1, PERIOD, [("1002", 3_000_000, 0), ("3001", 0, 2_000_000)])
_v(db, 1, PERIOD, [("1002", 300_000, 0), ("3002", 0, 300_000)])
_v(db, 1, PERIOD, [("1002", 200_000, 0), ("3101", 0, 200_000)])
_v(db, 1, PERIOD, [("3103", 0, 14_918_101.66), ("3104", 5_985_512.16, 0)])
seed_opening(db, 1, [("1002", "银行存款", "debit", 0)])
p = _analyze(db)
items = p["items"]
assert ("3001", 1) in items["4001"]["codes"] # 实收资本（漏 3001 = 漏 200 万）
assert ("3002", 1) in items["4002"]["codes"] # 资本公积
assert ("3101", 1) in items["4101"]["codes"] # 盈余公积
assert ("3103", 1) in items["4103|4104"]["codes"] and ("3105", 1) in items["4103|4104"]["codes"]
# 权益合计 = 2,000,000 + 300,000 + 200,000 + 14,918,101.66 − 5,985,512.16
assert p["totals_after"]["equity"] == pytest.approx(11_432_589.50)
lines = {k: v for k, v in p["after"].items()}
assert lines["4001"] == pytest.approx(2_000_000.0) # 实收资本不再为 0 / 不漏
def test_t15_opening_is_entity_scoped(self, db: Session):
"""期初层按账套隔离：他账套期初不串账；无期初账套 = 不可信"""
seed_opening(db, 1, [("1002", "银行存款", "debit", 500_000)])
_v(db, 2, PERIOD, [("1002", 9_999, 0), ("4001", 0, 9_999)])
p1 = _analyze(db, entity_id=1)
p2 = _analyze(db, entity_id=2)
assert p1["invariants"]["opening_imported"] is True
assert p2["invariants"]["opening_imported"] is False
assert p2["direction_check_unreliable"] is True # entity2 期初未补 → 不可信
assert p2["raw"]["1001|1002|1012"] == pytest.approx(9_999.0) # 未被 entity1 期初污染