diff --git a/backend/scripts/kpi_value_collector.py b/backend/scripts/kpi_value_collector.py index c3772242..0b31871f 100644 --- a/backend/scripts/kpi_value_collector.py +++ b/backend/scripts/kpi_value_collector.py @@ -27,7 +27,8 @@ if __name__ == "__main__": import os sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) -from sqlalchemy import func +from sqlalchemy import and_, func, not_, or_ +from sqlalchemy.orm import aliased from app.database import get_session_local from app.models import ( KPIValueSource, KPIValueCollectLog, KPIValue, KPIDefinition, @@ -71,6 +72,13 @@ def collect_for_mapping(db, mapping: KPIValueSource, period: str, write_kpi: boo period_field = mapping.period_field or "period" unit = mapping.unit_conversion or 1 + # 现金流两族KPI 激活(2026-09-14): 现金流量表直接法·对手方分类口径 + # filter_rule 形如 {"calc":"cashflow_direct","bucket":"operating","cash_subjects":["1001%","1002%","1012%"], + # "financing_counterpart":[...],"investing_counterpart":[...],"round":4, + # "minus_capex":{"counterpart":[...],"side":"debit"}} + if isinstance(filter_rule, dict) and filter_rule.get("calc") == "cashflow_direct": + return collect_cashflow_direct(db, mapping, period, write_kpi=write_kpi) + # P1③④ 激活(2026-09-14): 科目组合映射 —— 由 kpi_subject_map 推导的多科目加权归集 # filter_rule 形如 {"subject_codes":["5001%"],"measure":"credit","weights":{"5001%":1.0}, # "calc":"sum|ratio","denominator":"5001%","scale":100} @@ -258,6 +266,118 @@ def collect_subject_group(db, mapping, period: str, write_kpi: bool = True): return value, message +# ── 现金流量表直接法·对手方分类归集（2026-09-14 激活）────────────────── +# 任务: yanxue-budget-cashflow-kpi-activate-20260914（财务Bot 口径说明 §二/§三/§五） +# 口径: 以**凭证为单元**取含现金类科目(1001%/1002%/1012%)的凭证 → 对手方**行级**分类 +# (筹资类/投资类/其余=经营)；经营净额 = Σ(经营类对手方行净额 × -1) +# 恒等式: 经营净额 + 筹资净额 + 投资净额 = 现金总净额 +# 关键: 必须行级拆分，不能按凭证级分类 —— 反例 2026-06 凭证 31226 = 还借款 2,550,000 +# + 手续费用 18,700 同单，整单归筹资会把 18,700 手续费误算作经营（高估 18,700）。 +# 纪律: 不硬编码凭证号/常量；对手方集合、bucket、舍位全部来自 filter_rule。 +_DEFAULT_CASH_SUBJECTS = ("1001%", "1002%", "1012%") +_CASHFLOW_BUCKETS = ("operating", "financing", "investing", "net_cash") + + +def _cashflow_voucher_subquery(db, entity_id: int, period: Optional[str], cash_subjects): + """含现金类科目的凭证键集合（entity_id + voucher_date + voucher_no + period）。""" + q = db.query( + VoucherDetail.entity_id.label("entity_id"), + VoucherDetail.voucher_date.label("voucher_date"), + VoucherDetail.voucher_no.label("voucher_no"), + VoucherDetail.period.label("period"), + ).filter(VoucherDetail.entity_id == entity_id) + if period: + q = q.filter(VoucherDetail.period == period) + q = q.filter(or_(*[VoucherDetail.subject_code.like(p) for p in cash_subjects])) + voucher_cnt = q.distinct().count() + return q.distinct().subquery(), voucher_cnt + + +def _cashflow_side_amount(db, entity_id: int, period: Optional[str], cash_subjects, + include_prefixes=None, exclude_prefixes=None, side: str = "cashflow"): + """现金凭证内「对手方行级」金额 → (金额元, 行数, 凭证数)。 + + side="cashflow" : Σ(贷方 − 借方)，即现金流口径（流入为正） + side="debit_net" : Σ(借方 − 贷方)，即资本支出/借方净额口径（付现为正） + include_prefixes / exclude_prefixes: 4 位科目前缀，命中即纳入 / 排除（均为 LIKE 前缀，非硬编码）。 + """ + sub, voucher_cnt = _cashflow_voucher_subquery(db, entity_id, period, cash_subjects) + c = aliased(VoucherDetail) + if side == "debit_net": + expr = func.coalesce(func.sum(c.debit_amount - c.credit_amount), 0) + else: + expr = func.coalesce(func.sum(c.credit_amount - c.debit_amount), 0) + q = db.query(expr, func.count(c.id)).select_from(c).join( + sub, + and_(sub.c.entity_id == c.entity_id, sub.c.voucher_date == c.voucher_date, + sub.c.voucher_no == c.voucher_no, sub.c.period == c.period), + ).filter(c.entity_id == entity_id) + if period: + q = q.filter(c.period == period) + # 只保留对手方行（剔除现金类科目自身） + q = q.filter(not_(or_(*[c.subject_code.like(p) for p in cash_subjects]))) + if include_prefixes: + q = q.filter(or_(*[c.subject_code.like(str(p) + "%") for p in include_prefixes])) + if exclude_prefixes: + q = q.filter(not_(or_(*[c.subject_code.like(str(p) + "%") for p in exclude_prefixes]))) + row = q.first() + amount = float(row[0] or 0) if row else 0.0 + rows = int(row[1] or 0) if row else 0 + return amount, rows, voucher_cnt + + +def collect_cashflow_direct(db, mapping, period: str, write_kpi: bool = True): + """现金流量表直接法归集（filter_rule.calc = cashflow_direct），返回 (采集值, 说明)。 + + bucket: operating(经营/默认) | financing(筹资) | investing(投资) | net_cash(现金总净额) + minus_capex: 付现资本支出扣减项 {"counterpart":[4位前缀...],"side":"debit"} + """ + rule = mapping.filter_rule or {} + cash_subjects = [str(p) for p in (rule.get("cash_subjects") or _DEFAULT_CASH_SUBJECTS)] + financing = [str(p) for p in (rule.get("financing_counterpart") or [])] + investing = [str(p) for p in (rule.get("investing_counterpart") or [])] + bucket = str(rule.get("bucket") or "operating") + dp = int(rule.get("round", 4) or 4) + unit = mapping.unit_conversion or 1 + + if bucket not in _CASHFLOW_BUCKETS: + return None, f"未知 bucket={bucket}（可选 {'/'.join(_CASHFLOW_BUCKETS)}）" + + if bucket == "net_cash": + amount, rows, vouchers = _cashflow_side_amount(db, mapping.entity_id, period, cash_subjects) + elif bucket == "financing": + amount, rows, vouchers = _cashflow_side_amount(db, mapping.entity_id, period, cash_subjects, + include_prefixes=financing) + elif bucket == "investing": + amount, rows, vouchers = _cashflow_side_amount(db, mapping.entity_id, period, cash_subjects, + include_prefixes=investing) + else: + amount, rows, vouchers = _cashflow_side_amount(db, mapping.entity_id, period, cash_subjects, + exclude_prefixes=financing + investing) + + if vouchers == 0 or rows == 0: + return None, f"源表无记录(现金流量直接法 period={period}, bucket={bucket})" + + value_yuan = amount + parts = [f"{bucket} {amount:.2f}元"] + capex_rule = rule.get("minus_capex") or {} + if capex_rule: + capex_prefixes = [str(p) for p in (capex_rule.get("counterpart") or [])] + capex_side = "debit_net" if str(capex_rule.get("side") or "debit").startswith("debit") else "cashflow" + capex, capex_rows, _ = _cashflow_side_amount(db, mapping.entity_id, period, cash_subjects, + include_prefixes=capex_prefixes, side=capex_side) + value_yuan = amount - capex + parts.append(f"资本支出({capex_side}) {capex:.2f}元/{capex_rows}行") + + value = round(float(value_yuan) * unit, dp) + message = (f"现金流量直接法(period={period}, bucket={bucket}, 现金凭证{vouchers}张/对手方{rows}行): " + + " − ".join(parts) + f" × {unit}") + if write_kpi: + _upsert_auto_value(db, mapping, period, value, + f"自动归集: 现金流量直接法({bucket})") + return value, message + + def _batch_no() -> str: return f"auto-{datetime.now().strftime('%Y%m%d%H%M%S')}" diff --git a/backend/scripts/verify_cashflow_direct_caliber.py b/backend/scripts/verify_cashflow_direct_caliber.py new file mode 100644 index 00000000..43d44a4f --- /dev/null +++ b/backend/scripts/verify_cashflow_direct_caliber.py @@ -0,0 +1,130 @@ +#!/usr/bin/env python3 +"""回归电池: 现金流量表直接法口径（cashflow_direct）—— KPI4 F_OP_CFLOW / KPI45 F_FCF。 + +任务: yanxue-budget-cashflow-kpi-activate-20260914（财务口径说明 §二/§三/§五 落地复核） + +判据（全部只读，write_kpi=False）: + 1. 生产同一代码路径试算 2026-01..08，与现网 zxt-sync 权威值对账（容差 0.001） + —— 2026-08 因上游已撤单凭证仍在库内（智享通快照不含）按已知差异放行（WARN，不算 FAIL） + 2. 恒等式: 经营净额 + 筹资净额 + 投资净额 = 现金总净额（每期 |Δ| ≤ 0.001） + 3. 负例对照: 若按「凭证级整单分类」（错误做法），2026-06 应为 453,697.63 元 + （行级 = 434,997.63 元，差 +18,700 = 凭证 31226 同单内的银行手续费）→ 证明必须行级拆分 + 4. F_FCF = 经营净额 − 资本支出；酣客 2026 年 capex=0 → 与经营净额同值 + +用法: /usr/bin/python3 backend/scripts/verify_cashflow_direct_caliber.py （退出码 0 = 全部通过） +""" +import os +import sys + +sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) +from sqlalchemy import text +from app.database import get_session_local +from app.models import KPIValueSource, KPIValue +from scripts.kpi_value_collector import collect_for_mapping + +ENTITY = 1 +PERIODS = ["2026-0%d" % i for i in range(1, 9)] +TOL = 0.001 +KNOWN_DIFF_PERIODS = {"2026-08"} # 上游已撤单凭证未标记 → 已知差异（挂账，非回归） +RULE = { + "calc": "cashflow_direct", "measure": "net", "version": 1, "bucket": "operating", + "cash_subjects": ["1001%", "1002%", "1012%"], + "financing_counterpart": ["2001", "2501", "2502", "3001", "3002", "3101", "3103", "3104", "2232", "2241"], + "investing_counterpart": ["1501", "1511", "1601", "1602", "1603", "1604", "1605", "1606", "1701", "1702"], + "round": 4, +} +CAPEX = {"counterpart": ["1601", "1604", "1605", "1606", "1701", "1702"], "side": "debit"} +CASH_PREFIX = ("1001", "1002", "1012") +FIN = tuple(RULE["financing_counterpart"]) +INV = tuple(RULE["investing_counterpart"]) + +fails, warns = [], [] + + +def probe(db, kpi_id, period, rule): + m = KPIValueSource(entity_id=ENTITY, kpi_id=kpi_id, source_table="voucher_details", + source_field="credit_amount", aggregate="sum", filter_rule=rule, + period_field="period", unit_conversion=0.0001, status="active") + return collect_for_mapping(db, m, period, write_kpi=False) + + +def ref(db, kpi_id, period): + row = db.query(KPIValue).filter( + KPIValue.kpi_id == kpi_id, KPIValue.period == period, + KPIValue.source_type == "zxt-sync", KPIValue.actual_value.isnot(None), + ).order_by(KPIValue.calculated_at.desc()).first() + return float(row.actual_value) if row else None + + +def main(): + db = get_session_local()() + try: + print("== 判据1: 8 期对账（现网 zxt-sync）==") + for p in PERIODS: + for kpi in (4, 45): + rule = dict(RULE, minus_capex=CAPEX) if kpi == 45 else dict(RULE) + val, _ = probe(db, kpi, p, rule) + r = ref(db, kpi, p) + ok = val is not None and r is not None and abs(val - r) <= TOL + tag = "PASS" if ok else ("WARN" if p in KNOWN_DIFF_PERIODS else "FAIL") + print(f" {tag} KPI{kpi} {p}: 算={val} 现网={r}") + if not ok: + (warns if p in KNOWN_DIFF_PERIODS else fails).append(f"KPI{kpi} {p}") + + print("\n== 判据2: 恒等式（经营+筹资+投资=现金总净额）& 判据4: FCF=capex扣减 ==") + for p in PERIODS: + op, _ = probe(db, 4, p, dict(RULE, bucket="operating")) + fin, _ = probe(db, 4, p, dict(RULE, bucket="financing")) + inv, _ = probe(db, 4, p, dict(RULE, bucket="investing")) + net, _ = probe(db, 4, p, dict(RULE, bucket="net_cash")) + fcf, _ = probe(db, 45, p, dict(RULE, minus_capex=CAPEX)) + f = lambda x: float(x) if x is not None else 0.0 + ident = abs(f(op) + f(fin) + f(inv) - f(net)) + same = abs(f(fcf) - f(op)) <= TOL # 酣客 capex=0 → FCF ≡ 经营净额 + ok = ident <= TOL and same + print(f" {'PASS' if ok else 'FAIL'} {p}: 经营={op} 筹资={fin} 投资={inv} 现金净额={net} " + f"identityΔ={round(ident, 4)} FCF={fcf}") + if not ok: + fails.append(f"identity/{p}") + + print("\n== 判据3: 负例对照（凭证级整单分类 = 错误做法）==") + sql = text(""" + SELECT v.voucher_no, v.subject_code, v.debit_amount, v.credit_amount + FROM voucher_details v + JOIN (SELECT DISTINCT entity_id, voucher_date, voucher_no, period FROM voucher_details + WHERE entity_id=:e AND period=:p + AND (subject_code LIKE '1001%' OR subject_code LIKE '1002%' OR subject_code LIKE '1012%')) w + ON w.entity_id=v.entity_id AND w.voucher_date=v.voucher_date + AND w.voucher_no=v.voucher_no AND w.period=v.period + WHERE v.entity_id=:e AND v.period=:p""") + rows = db.execute(sql, {"e": ENTITY, "p": "2026-06"}).fetchall() + line_op = vlevel_op = 0.0 + per_voucher = {} + for vno, code, dr, cr in rows: + code = str(code) + n = float(cr) - float(dr) + per_voucher.setdefault(vno, []).append((code, n)) + if code.startswith(CASH_PREFIX): + continue + if code[:4] not in FIN and code[:4] not in INV: + line_op += n + for vno, lines in per_voucher.items(): + cps = [(c, n) for c, n in lines if not c.startswith(CASH_PREFIX)] + if any(c[:4] in FIN or c[:4] in INV for c, _ in cps): + continue + vlevel_op += sum(n for _, n in cps) + ok = abs(line_op - 434997.63) <= 0.01 and abs(vlevel_op - line_op - 18700.0) <= 0.01 + print(f" {'PASS' if ok else 'FAIL'} 2026-06 行级={round(line_op, 2)} 凭证级={round(vlevel_op, 2)} " + f"Δ={round(vlevel_op - line_op, 2)}（预期行级 434,997.63 / Δ +18,700.00）") + if not ok: + fails.append("negative-control-2026-06") + + print(f"\n===== 汇总: {'全部通过' if not fails else 'FAIL ' + str(len(fails))} " + f"| WARN(已知差异) {len(warns)} 项 {warns if warns else ''} =====") + return 1 if fails else 0 + finally: + db.close() + + +if __name__ == "__main__": + sys.exit(main())