feat(cashflow): 期初/期末现金补采 — 科目余额表货币资金(1001/1002/1012) 入库

- import_cashflow_statements.py 新增 _read_cash_balances(): 读 <YYYY.MM>/kmyeb.csv 货币资金(1001/1002/1012) 期初借-贷 / 期末借-贷, ÷10000 留 2 位 → begin_cash/end_cash - 仅 entity_id=1(智享通通道=酣客) 生效, 其它账套保持 NULL; 源缺失/科目缺失保持 NULL 不写 0 - 既有三大活动净额与 remark NA 固定短语逻辑未动; dry-run 默认/--period/幂等语义不变 - --self-test 扩至 9/9: 新增 kmyeb缺失/科目缺失/借贷为负三类边界 + 真实源自洽与跨期连续
2026-09-23 16:20:43 +08:00
parent e79b51666d
commit d13bf1b6e4
1 changed files with 140 additions and 8 deletions
@@ -10,6 +10,16 @@
     列：分类,编码,项目名称,方向,金额     分类∈{经营活动,投资活动,筹资活动}
  兼容 2026.08 的另存名 cashflow_stat_<YYYY.MM>.csv

+期初/期末现金数据源（本版本新增，任务 project-20260923-cashflow-beginend-cash-backfill）：
+  /root/WB_WS/_data/zxt-hanke/<YYYY.MM>/kmyeb.csv（科目余额表，只读）
+     取 科目编码 ∈ {1001 库存现金, 1002 银行存款, 1012 其他货币资金}
+     期初 = Σ(期初借 − 期初贷)；期末 = Σ(期末借 − 期末贷)；元 → 万元（÷10000，2 位）
+     依据：科目余额表货币资金逐期变动 == 库内三大活动净额合计（8/8 期吻合 ±0.01，
+     2026-09-23 交叉验证）；上游「现金流量表主表」（ReportDesign 1405）金额实测全 0 不可用
+     ⇒ 主表口径（历史 §项1-② 缺陷）不改，改走科目余额表货币资金。
+     该源属**智享通通道 = 酣客账套**（KMYEB_ENTITY_ID），故仅对 entity_id=1 生效；
+     其它账套保持 NULL（不得把酣客货币资金写到别家账套）。
+
 口径：
  - 经营/投资/筹资 净额 = Σ(流入) − Σ(流出)，单位 元 → 万元（÷10000，保留 2 位）
  - **NA（无发生额）标注（本版本新增，裁定 finance-20260916-cma-caliber-ruling §2.1）**：
@@ -18,13 +28,16 @@
    供判定层用 `parse_na_activities(remark)` 取回 → **仅作标注与机读入口**；
    口径裁定（2026-09-17）已定：该维按 **0 如实参与判定**（八类判定的**二维映射**，
    经营(±)×筹资(±) 出 2 个候选类型），**不再**因 NA/0 整体判「未判定」。
-  - source='official'（官方报表口径）；期初/期末现金 留空（主表采集缺陷，见报告 §项1-②）
+  - source='official'（官方报表口径）
+  - **期初/期末现金**（本版本新增）：取科目余额表货币资金（1001/1002/1012）→ 万元 2 位，
+    源文件/科目缺失时保持 NULL（不写 0 造数）；仅 entity_id=1（智享通通道=酣客）生效
  - 同 (entity_id, period) 重复执行 = 覆盖更新（幂等）

 用法：
  python3 scripts/import_cashflow_statements.py                       # dry-run（默认，只算不写）
  python3 scripts/import_cashflow_statements.py --period 2026-07      # 单期 dry-run
-  python3 scripts/import_cashflow_statements.py --self-test           # NA 标注逻辑自测（零库连接）
+  python3 scripts/import_cashflow_statements.py --self-test           # 标注+期初期末自测（零库连接；
+                                                                      # 边界用例只在系统临时目录造样，不碰生产源）
  python3 scripts/import_cashflow_statements.py --apply               # 落库（需放行）
 """
 import argparse
@@ -33,6 +46,7 @@ import glob
 import os
 import re
 import sys
+import tempfile

 sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))

@@ -43,6 +57,13 @@ from app.api.cashflow_structure import classify_enterprise
 ZKT_BASE = "/root/WB_WS/_data/zxt-hanke/cashflow"
 CLASSES = ("经营活动", "投资活动", "筹资活动")

+#: 科目余额表（kmyeb.csv）根目录 —— 与 cashflow/ 同级，期初/期末现金来源
+KMYEB_BASE = "/root/WB_WS/_data/zxt-hanke"
+#: 货币资金科目：库存现金 / 银行存款 / 其他货币资金
+CASH_CODES = ("1001", "1002", "1012")
+#: kmyeb 属智享通通道 = 酣客账套；其它账套不得套用（防跨账套写脏）
+KMYEB_ENTITY_ID = 1
+
 #: NA 固定短语模板 —— 判定层与量纲守卫共用，勿改字面量（裁定 §2.1 指定）
 NA_PHRASE_TMPL = "{cls}本期无发生额（无项目行）"

@@ -93,6 +114,52 @@ def _nets(rows):
    return {c: round(v["in"] - v["out"], 2) for c, v in agg.items()}, agg


+def _read_cash_balances(period, base=KMYEB_BASE):
+    """读科目余额表货币资金 → (begin_wan, end_wan, meta)。
+
+    期初 = Σ(期初借 − 期初贷)、期末 = Σ(期末借 − 期末贷)，科目 ∈ CASH_CODES，元 → 万元 2 位。
+    借方−贷方为**有符号**口径：贷方大于借方即为负（红字/透支/方向反），**不得取绝对值或清零**。
+    源文件缺失 / 无任一货币资金科目行 ⇒ 返回 (None, None, meta)，由调用方保持 NULL（不写 0 造数）。
+    """
+    day_dir = os.path.join(base, period.replace("-", "."))
+    src = os.path.join(day_dir, "kmyeb.csv")
+    if not os.path.exists(src):
+        return None, None, {"status": "SKIP_no_kmyeb", "src": src, "codes": []}
+    beg = 0.0
+    end = 0.0
+    hits = []
+    with open(src, encoding="utf-8-sig") as fh:
+        for r in csv.DictReader(fh):
+            code = (r.get("科目编码") or "").strip()
+            if code not in CASH_CODES:
+                continue
+            try:
+                b = float((r.get("期初借") or "0").strip() or 0) - float((r.get("期初贷") or "0").strip() or 0)
+                e = float((r.get("期末借") or "0").strip() or 0) - float((r.get("期末贷") or "0").strip() or 0)
+            except ValueError:
+                continue
+            hits.append(code)
+            beg += b
+            end += e
+    if not hits:
+        return None, None, {"status": "SKIP_no_cash_subject", "src": src, "codes": []}
+    return (round(beg / 10000.0, 2), round(end / 10000.0, 2),
+            {"status": "READY", "src": src, "codes": hits})
+
+
+def _write_kmyeb(root, period, rows):
+    """自测专用：在临时目录造 kmyeb.csv 样件（表头与生产源同构）。"""
+    d = os.path.join(root, period.replace("-", "."))
+    os.makedirs(d, exist_ok=True)
+    p = os.path.join(d, "kmyeb.csv")
+    with open(p, "w", encoding="utf-8-sig", newline="") as fh:
+        w = csv.writer(fh)
+        w.writerow(["科目编码", "科目名称", "期初借", "期初贷", "本期借", "本期贷",
+                    "本年借", "本年贷", "当前借", "当前贷", "期末借", "期末贷"])
+        w.writerows(rows)
+    return p
+
+
 def collect(period_filter=None):
    """扫描盘上各期 → [payload dict]（万元，未写库）"""
    payloads = []
@@ -122,6 +189,7 @@ def collect(period_filter=None):
            os.path.basename(src), len(rows))
        remark = (head + "；NA(无发生额，按0参与判定): " + "、".join(na_phrases)) if na_phrases \
            else (head + "；三类活动均有项目行（无 NA）")
+        cash_begin, cash_end, cash_meta = _read_cash_balances(period)
        payloads.append({
            "period": period,
            "status": "READY",
@@ -134,6 +202,9 @@ def collect(period_filter=None):
            "na_phrases": na_phrases,
            "na_flags": {c: (c in missing) for c in CLASSES},
            "remark": remark,
+            "begin_cash": cash_begin,
+            "end_cash": cash_end,
+            "cash_meta": cash_meta,
        })
    return payloads

@@ -163,12 +234,58 @@ def _self_test():
    assert ready, "无可入库期（数据源缺失）"
    all_inv_na = all("投资活动" in p["na_classes"] for p in ready)
    passed += 1
-    print("=== NA 标注 self-test: %d/5 PASS | 期数=%d | 投资活动全NA=%s ==="
-          % (passed, len(ready), all_inv_na))
-    for p in ready:
-        print("  %-8s NA=%s" % (p["period"], "、".join(p["na_classes"]) or "无"))
-        print("           remark: %s" % p["remark"])
    assert all_inv_na, "酣客 8 期投资活动应全为 NA（裁定 §2.1 实证）"
+
+    # 6~8) 期初/期末现金取数边界（临时目录造样，零生产源写）
+    with tempfile.TemporaryDirectory() as tmp:
+        # 6) kmyeb 缺失 → 保持 NULL（不得写 0）
+        b6, e6, m6 = _read_cash_balances("2026-01", base=tmp)
+        assert (b6, e6) == (None, None) and m6["status"] == "SKIP_no_kmyeb", (b6, e6, m6)
+        passed += 1
+        # 7) 文件在但无货币资金科目（非现金科目不得混入）
+        _write_kmyeb(tmp, "2026-01", [
+            ["1122", "应收账款", "999", "0", "0", "0", "0", "0", "0", "0", "999", "0"]])
+        b7, e7, m7 = _read_cash_balances("2026-01", base=tmp)
+        assert (b7, e7) == (None, None) and m7["status"] == "SKIP_no_cash_subject", (b7, e7, m7)
+        passed += 1
+        # 8) 借贷为负（贷>借）：负号必须保留，不得 abs/清零；非现金科目应被忽略
+        _write_kmyeb(tmp, "2026-02", [
+            ["1001", "库存现金", "0", "5000", "0", "0", "0", "0", "0", "0", "0", "12000"],
+            ["1002", "银行存款", "20000", "0", "0", "0", "0", "0", "0", "0", "30000", "0"],
+            ["1122", "应收账款", "999", "0", "0", "0", "0", "0", "0", "0", "999", "0"]])
+        b8, e8, m8 = _read_cash_balances("2026-02", base=tmp)
+        assert (b8, e8) == (1.5, 1.8), (b8, e8)
+        assert m8["status"] == "READY" and sorted(m8["codes"]) == ["1001", "1002"], m8
+        _write_kmyeb(tmp, "2026-03", [
+            ["1001", "库存现金", "0", "12000", "0", "0", "0", "0", "0", "0", "0", "12000"]])
+        b8b, e8b, _m = _read_cash_balances("2026-03", base=tmp)
+        assert (b8b, e8b) == (-1.2, -1.2), (b8b, e8b)
+        passed += 1
+
+    # 9) 真实源 8 期：期初/期末齐全 + 与三大净额自洽（期末−期初 == Σ净额 ±0.02）+ 跨期连续（±0.01）
+    with_cash = [p for p in ready
+                 if p["begin_cash"] is not None and p["end_cash"] is not None]
+    assert len(with_cash) == len(ready), "部分期未取到货币资金：%d/%d" % (len(with_cash), len(ready))
+    for p in ready:
+        s = round(sum(p["nets_wan"][c] for c in CLASSES), 2)
+        d = round(p["end_cash"] - p["begin_cash"], 2)
+        assert abs(d - s) <= 0.02, ("期末−期初≠净额合计", p["period"], d, s)
+    prev_end = None
+    for p in ready:
+        if prev_end is not None:
+            assert abs(p["begin_cash"] - prev_end) <= 0.01, (
+                "跨期不连续", p["period"], p["begin_cash"], prev_end)
+        prev_end = p["end_cash"]
+    passed += 1
+
+    print("=== self-test: %d/9 PASS | 期数=%d | 投资活动全NA=%s | 期初/期末齐全=%d/%d ==="
+          % (passed, len(ready), all_inv_na, len(with_cash), len(ready)))
+    for p in ready:
+        print("  %-8s NA=%s | 期初=%.2f 期末=%.2f 万（Δ=%.2f / 净额合计=%.2f）"
+              % (p["period"], "、".join(p["na_classes"]) or "无", p["begin_cash"], p["end_cash"],
+                 round(p["end_cash"] - p["begin_cash"], 2),
+                 round(sum(p["nets_wan"][c] for c in CLASSES), 2)))
+        print("           remark: %s" % p["remark"])
    return 0


@@ -218,6 +335,14 @@ def main():
                     verdict, os.path.basename(p["src"])))
            print("           NA(无发生额)=%s；投资栏标注=%s" % (
                "、".join(p["na_classes"]) or "无", cls.get("investing_note") or "-"))
+            cash_applicable = (args.entity_id == KMYEB_ENTITY_ID)
+            has_cash = p["begin_cash"] is not None and p["end_cash"] is not None
+            print("           期初/期末现金 = %s / %s 万（源 %s）%s" % (
+                "NULL" if p["begin_cash"] is None else "%.2f" % p["begin_cash"],
+                "NULL" if p["end_cash"] is None else "%.2f" % p["end_cash"],
+                os.path.basename(p["cash_meta"]["src"]),
+                "" if cash_applicable else
+                "；账套 %d 非智享通通道 → 不写期初/期末" % args.entity_id))
            if not args.apply:
                continue
            rec = db.query(CashFlowStructureValue).filter(
@@ -233,9 +358,16 @@ def main():
            rec.financing_net = n["筹资活动"]
            rec.source = "official"
            rec.remark = p["remark"]
+            if cash_applicable and has_cash:
+                # 期初/期末现金：科目余额表货币资金（已 ÷10000 留 2 位）
+                rec.begin_cash = p["begin_cash"]
+                rec.end_cash = p["end_cash"]
            db.flush()
            written += 1
-            print("      → %s (id=%s)" % ("入库" if created else "更新", rec.id))
+            print("      → %s (id=%s) 期初/期末=%s" % (
+                "入库" if created else "更新", rec.id,
+                "NULL（源缺失，保持原值）" if not (cash_applicable and has_cash)
+                else "%.2f / %.2f 万" % (rec.begin_cash, rec.end_cash)))
        if args.apply:
            db.commit()
            print("=== 已提交 %d 期 ===" % written)