refactor#每单链路、清算核对增加明细项。

code/post_clear_check/src/post_clear_check/chain.py

@@ -103,6 +103,7 @@ DIMENSION_NAME = {
 # 订单卡上的索引和批次状态。不参与总结论，读回后仍靠这个前缀识别。
 INDEX_MARK = "关联报表索引："
 STATUS_MARK = "批次状态："
+CLEAR_MARK = "清算对照："
 # 内部链路可以出现在哪些方向上。收卡按产生来源，这张表只用来确认没有链路被漏掉。
 CHAIN_DIMENSIONS = {
     CHAIN_BUY: frozenset({DIM_ORDER, DIM_ISSUE, DIM_SALES}),

code/post_clear_check/src/post_clear_check/dimension_body.py

@@ -17,6 +17,7 @@ from post_clear_check.chain import (
     DIM_TRS,
     DIMENSION_NAME,
     GATE_MISSING_TABLE,
+    CLEAR_MARK,
     INDEX_MARK,
     STATUS_MARK,
 )
@@ -120,11 +121,13 @@ def _body(chain_name: str, views: tuple[_View, ...], business_date: date) -> lis
         for view in views
         if not view.formula_text.startswith(INDEX_MARK)
         and not view.formula_text.startswith(STATUS_MARK)
+        and not view.formula_text.startswith(CLEAR_MARK)
         and view.formula_text not in GATE_FORMULAS
     ]
+    clears = [view.calc_text for view in views if view.formula_text.startswith(CLEAR_MARK) and view.calc_text]
     pieces = [_summary(chain_name, views)]
     if chain_name == DIM_ORDER:
-        pieces.extend(_order_sections(rest))
+        pieces.extend(_order_sections(rest, clears))
         pieces.append(_index_chapter(annotations))
     else:
         if gates:
@@ -145,6 +148,7 @@ def _summary(chain_name: str, views: tuple[_View, ...]) -> str:
         for view in views
         if not view.formula_text.startswith(INDEX_MARK)
         and not view.formula_text.startswith(STATUS_MARK)
+        and not view.formula_text.startswith(CLEAR_MARK)
         and view.formula_text not in GATE_FORMULAS
     ]
     buckets: dict[str, str] = {}
@@ -179,7 +183,7 @@ def _summary(chain_name: str, views: tuple[_View, ...]) -> str:
     return "\n".join(rows)
 
 
-def _order_sections(rows: Sequence[_View]) -> list[str]:
+def _order_sections(rows: Sequence[_View], clears: Sequence[str]) -> list[str]:
     """每单链路只写总结。盘中公式和清算对照按字段写五行，字段之间空一行。"""
 
     grouped: dict[str, list[_View]] = {code: [] for code, _title in _ORDER_SECTIONS}
@@ -192,6 +196,8 @@ def _order_sections(rows: Sequence[_View]) -> list[str]:
             found.append(_tight(title, [_link_summary(section_rows)]))
             continue
         blocks = _field_blocks(section_rows)
+        if code == "clear":
+            blocks.extend(clears)
         found.append(_tight(title, blocks) if blocks else _tight(title, ["本轮没有这类结论。"]))
     return found
 
@@ -298,7 +304,7 @@ def _field_blocks(rows: Sequence[_View]) -> list[str]:
         blocks.append(
             _lines(
                 (
-                    f"字段：{formula}",
+                    f"字段：{_clear_title(formula)}",
                     f"期望：{_expect(sample) if verdict != VERDICT_CORRECT else formula}",
                     f"实据：{actual}",
                     f"依据：{basis}",
@@ -421,7 +427,68 @@ def _index_chapter(lines: Sequence[_View]) -> str:
     return _tight("关联报表", blocks)
 
 
+def format_topic(title: str, lines: Sequence[object]) -> str:
+    """把一组已有结论收成巡检稿的五行。没有结论时写明未核，不编造金额。"""
+
+    views = [_View(line) for line in lines]
+    if not views:
+        return _lines(
+            (
+                f"字段：{title}",
+                "期望：该项已登记的公式",
+                "实据：本轮结论行没有这项核对",
+                "依据：无",
+                "结论：本轮未核",
+            )
+        )
+    verdict = VERDICT_CORRECT
+    for view in views:
+        verdict = _worse(verdict, view.verdict)
+    orders = _worst_orders(views)
+    counts = {VERDICT_CORRECT: 0, VERDICT_ERROR: 0, VERDICT_SUSPECT: 0}
+    for item in orders.values():
+        counts[item] += 1
+    names = _formula_names(views)
+    bases = _unique(view.basis for view in views if view.basis)
+    sample = next((view.calc_text for view in views if view.calc_text and view.calc_text not in {"一致", title}), "")
+    if len(sample) > 180:
+        sample = sample[:180] + "…"
+    label = {VERDICT_CORRECT: "正确", VERDICT_ERROR: "错误", VERDICT_SUSPECT: "可疑"}[verdict]
+    actual = (
+        f"{title}：本轮主单 {len(orders)} 笔，"
+        f"正确 {counts[VERDICT_CORRECT]}，"
+        f"错误 {counts[VERDICT_ERROR]}，"
+        f"可疑 {counts[VERDICT_SUSPECT]}。"
+        f"列核对 {len(views)} 条。"
+    )
+    if sample:
+        actual += f"例：{sample}"
+    return _lines(
+        (
+            f"字段：{title}",
+            _stacked("期望", names[:8] or ["该项已登记的公式"]),
+            f"实据：{actual}",
+            _stacked("依据", bases[:6] or ["见各列登记"]),
+            f"结论：{label}（{counts[VERDICT_CORRECT]}/{len(orders)}）",
+        )
+    )
+
+
+def _clear_title(formula: str) -> str:
+    """清算对照里把已有公式收成巡检稿的字段名。"""
+
+    if "已发布结果" in formula:
+        return "已发布订单执行列"
+    if "工作快照" in formula:
+        return "T日订单结算与工作快照"
+    if "当前投影" in formula:
+        return "当前投影"
+    return formula
+
+
 def _index_block(text: str) -> str:
+    if text.startswith("字段："):
+        return text
     head, _, rest = text.partition("：")
     actual = rest.split("。", 1)[0] if rest else text
     detail = "对应报表卡"

code/post_clear_check/src/post_clear_check/orchestrate.py

@@ -59,6 +59,7 @@ from post_clear_check.chain import (
     DIMENSIONS,
     DIMENSION_NAME,
     GATE_MISSING_TABLE,
+    CLEAR_MARK,
     INDEX_MARK,
     OWNER_BATCH,
     OWNER_FUND,
@@ -71,7 +72,7 @@ from post_clear_check.chain import (
     owner_of,
     unassign,
 )
-from post_clear_check.dimension_body import GATE_FORMULAS, dimension_of_gate
+from post_clear_check.dimension_body import GATE_FORMULAS, dimension_of_gate, format_topic
 from post_clear_check.issue_common import is_population_bucket
 from post_clear_check.issue_reports import check_issue_reports
 from post_clear_check.order_cell import CheckLine
@@ -346,10 +347,11 @@ def assemble_cards(
         buckets[dimension_of_gate(line.formula_text, line.check_text)].append(line)
     if buckets[DIM_ORDER]:
         buckets[DIM_ORDER] = [
-            *_status_lines(buckets),
-            *buckets[DIM_ORDER],
-            *_index_lines(population, buckets),
-        ]
+        *_status_lines(buckets),
+        *buckets[DIM_ORDER],
+        *_clear_topic_lines(buckets, population),
+        *_index_lines(population, buckets),
+    ]
     drafts: list[CardDraft] = []
     for code in DIMENSIONS:
         rows = tuple(buckets[code])
@@ -385,7 +387,11 @@ def _table_dimension(table: str) -> str:
 
 
 def _annotation(line: CheckLine) -> bool:
-    return line.formula_text.startswith(INDEX_MARK) or line.formula_text.startswith(STATUS_MARK)
+    return (
+        line.formula_text.startswith(INDEX_MARK)
+        or line.formula_text.startswith(STATUS_MARK)
+        or line.formula_text.startswith(CLEAR_MARK)
+    )
 
 
 def _dimension_verdict(lines: Sequence[CheckLine]) -> str:
@@ -423,50 +429,134 @@ def _status_lines(buckets: Mapping[str, Sequence[CheckLine]]) -> list[CheckLine]
 
 
 def _index_lines(population: set[str], buckets: Mapping[str, Sequence[CheckLine]]) -> list[CheckLine]:
-    """本轮主单在三张报表卡上的对错笔数。没有主单号的账户级报表不收进来。"""
+    """关联报表按巡检稿展开。只收本轮主单挂得上的报表行。"""
 
     found: list[CheckLine] = []
-    for code, label in ((DIM_ISSUE, "发行"), (DIM_SALES, "销售"), (DIM_TRS, "TRS")):
-        grouped: dict[str, dict[str, str]] = {}
-        for line in buckets[code]:
-            if not line.report_code or line.master_order_no not in population:
-                continue
-            orders = grouped.setdefault(line.report_code, {})
-            orders[line.master_order_no] = _worse_verdict(orders.get(line.master_order_no), line.verdict)
-        for report_code in sorted(grouped):
-            orders = grouped[report_code]
-            counts = {VERDICT_CORRECT: 0, VERDICT_ERROR: 0, VERDICT_SUSPECT: 0}
-            for verdict in orders.values():
-                counts[verdict] += 1
-            bits = [f"{label} {report_code}：本轮主单 {len(orders)} 笔"]
-            for verdict, name in (
-                (VERDICT_CORRECT, "正确"),
-                (VERDICT_ERROR, "错误"),
-                (VERDICT_SUSPECT, "可疑"),
-            ):
-                if counts[verdict]:
-                    bits.append(f"{name} {counts[verdict]}")
-            text = "，".join(bits) + f"。明细在{DIMENSION_NAME[code]}。"
-            found.append(
-                CheckLine(
-                    formula_text=f"{INDEX_MARK}{report_code}",
-                    calc_text=text,
-                    calc_value=None,
-                    check_text=text,
-                    check_value=None,
-                    basis="",
-                    verdict=VERDICT_CORRECT,
-                    report_code=report_code,
-                )
+    for title, selected in _report_groups(buckets, population):
+        text = _clip(format_topic(title, selected), 1000)
+        found.append(
+            CheckLine(
+                formula_text=f"{INDEX_MARK}{title}",
+                calc_text=text,
+                calc_value=None,
+                check_text=title,
+                check_value=None,
+                basis="",
+                verdict=VERDICT_CORRECT,
+                # 标题如「发行 OTC007 / OTC008」超过 report_code 的 16 字，编号留空，标题在公式和正文里。
+                report_code="",
+            )
+        )
+    return found
+
+
+def _clear_topic_lines(buckets: Mapping[str, Sequence[CheckLine]], population: set[str]) -> list[CheckLine]:
+    """清算对照补上交收、流水、汇率和留痕。已发布执行列仍由订单自己的公式写出。"""
+
+    found: list[CheckLine] = []
+    for title, selected in _clear_groups(buckets, population):
+        text = _clip(format_topic(title, selected), 1000)
+        found.append(
+            CheckLine(
+                formula_text=f"{CLEAR_MARK}{title}",
+                calc_text=text,
+                calc_value=None,
+                check_text=title,
+                check_value=None,
+                basis="",
+                verdict=VERDICT_CORRECT,
             )
+        )
+    return found
+
+
+def _report_groups(
+    buckets: Mapping[str, Sequence[CheckLine]], population: set[str]
+) -> list[tuple[str, list[CheckLine]]]:
+    """巡检稿里的关联报表。发行和销售的 OTC004、OTC007、OTC008 各自成段。"""
+
+    specs = (
+        ("发行 OTC003", DIM_ISSUE, ("OTC003",)),
+        ("销售 OTC003", DIM_SALES, ("OTC003",)),
+        ("发行 OTC004", DIM_ISSUE, ("OTC004",)),
+        ("销售 OTC004", DIM_SALES, ("OTC004",)),
+        ("发行 OTC007 / OTC008", DIM_ISSUE, ("OTC007", "OTC008")),
+        ("销售 OTC007 / OTC008", DIM_SALES, ("OTC007", "OTC008")),
+    )
+    return [(title, _pick(buckets[code], population, reports)) for title, code, reports in specs]
+
+
+def _clear_groups(
+    buckets: Mapping[str, Sequence[CheckLine]], population: set[str]
+) -> list[tuple[str, list[CheckLine]]]:
+    """巡检稿清算对照里、不在订单公式上的四项。"""
+
+    issue = _touching(buckets[DIM_ISSUE], population)
+    sales = _touching(buckets[DIM_SALES], population)
+    trs = _touching(buckets[DIM_TRS], population)
+    return [
+        ("交收台账", [line for line in issue + sales if _ledger_line(line)]),
+        ("销售资金流水", [line for line in sales if _cash_line(line)]),
+        ("TRS 发生额与汇率口径", [line for line in trs if _fx_line(line)]),
+        ("清算核对留痕", [line for line in issue + trs if _trace_line(line)]),
+    ]
+
+
+def _pick(lines: Sequence[CheckLine], population: set[str], reports: tuple[str, ...]) -> list[CheckLine]:
+    return [line for line in _touching(lines, population) if line.report_code in reports]
+
+
+def _touching(lines: Sequence[CheckLine], population: set[str]) -> list[CheckLine]:
+    """主单号对上，或正文里写了本轮主单号。账户级空主单不整表拉进来。"""
+
+    found: list[CheckLine] = []
+    for line in lines:
+        if line.master_order_no in population:
+            found.append(line)
+            continue
+        blob = f"{line.check_text}\n{line.calc_text}"
+        if any(order_no and _mentions(blob, order_no) for order_no in population):
+            found.append(line)
     return found
 
 
-def _worse_verdict(current: str | None, verdict: str) -> str:
-    rank = {VERDICT_ERROR: 0, VERDICT_SUSPECT: 1, VERDICT_CORRECT: 2}
-    if current is None or rank[verdict] < rank[current]:
-        return verdict
-    return current
+def _mentions(blob: str, order_no: str) -> bool:
+    """主单号要整段命中。ORD-A 不能算进 ORD-A1。"""
+
+    start = 0
+    while True:
+        index = blob.find(order_no, start)
+        if index < 0:
+            return False
+        end = index + len(order_no)
+        if end >= len(blob) or not blob[end].isalnum():
+            return True
+        start = end
+
+
+def _ledger_line(line: CheckLine) -> bool:
+    text = line.formula_text
+    return text in {"交收推演", "T日交收结果", "销售交收"} or "交收金额" in text or text.startswith("交收日")
+
+
+def _cash_line(line: CheckLine) -> bool:
+    text = line.formula_text
+    return line.report_code == "OTC002" or text in {
+        "原始流水",
+        "T日流水",
+        "金额 = 同一流水编号的 fund_ledger 金额",
+        "金额 = 同一流水编号的流水金额，且流水业务类型、营业日期一致",
+    }
+
+
+def _fx_line(line: CheckLine) -> bool:
+    text = line.formula_text
+    return any(key in text for key in ("汇率", "发生金额", "发生额"))
+
+
+def _trace_line(line: CheckLine) -> bool:
+    blob = f"{line.formula_text}\n{line.check_text}\n{line.calc_text}"
+    return any(key in blob for key in ("留痕", "recon", "mismatch", "差异表"))
 
 
 def assemble_batch_cards(lines: Sequence[ConclusionLine]) -> tuple[CardDraft, ...]:
@@ -1276,6 +1366,7 @@ def _normalize_line(line: CheckLine) -> CheckLine:
         calc_text=_clip(line.calc_text, 1000),
         check_text=_clip(line.check_text, 1000),
         basis=_clip(line.basis, 500),
+        report_code=_clip(line.report_code, 16),
     )
 
 

code/post_clear_check/tests/test_feishu.py

@@ -15,6 +15,7 @@ from loguru import logger
 from pydantic import SecretStr
 
 from post_clear_check.chain import (
+    CLEAR_MARK,
     CARD_DIMENSION,
     CHAIN_BUY,
     CHAIN_CASH_CHANGE,
@@ -1390,6 +1391,45 @@ def test_link_summary_describes_facts_and_hedge_roots() -> None:
     assert "正确 8 笔，错误 0 笔，可疑 0 笔" not in body
 
 
+def test_order_card_expands_clearing_and_related_reports() -> None:
+    """清算对照和关联报表按巡检稿的字段展开，不把多列收成一行笔数。"""
+
+    body = render_body(
+        business_date=DAY,
+        trade_date=DAY,
+        batch_no="IB",
+        chain_name=DIM_ORDER,
+        record_no="14",
+        lines=(
+            _dim(
+                "T日订单结算与工作快照一致",
+                VERDICT_CORRECT,
+                master_order_no="ORD1",
+                calc="一致",
+                calc_value=None,
+                check_value=None,
+            ),
+            _dim(
+                f"{CLEAR_MARK}交收台账",
+                VERDICT_CORRECT,
+                calc="字段：交收台账\n期望：交收推演\n实据：交收台账：本轮主单 8 笔，正确 8，错误 0，可疑 0。\n依据：02 方案交收\n结论：正确（8/8）",
+                calc_value=None,
+                check_value=None,
+            ),
+            _dim(
+                "关联报表索引：发行 OTC003",
+                VERDICT_CORRECT,
+                calc="字段：发行 OTC003\n期望：名义本金 = 执行名义\n实据：发行 OTC003：本轮主单 8 笔，正确 8，错误 0，可疑 0。\n依据：产品原型 OTC003\n结论：正确（8/8）",
+                calc_value=None,
+                check_value=None,
+            ),
+        ),
+    )
+    assert "清算对照\n字段：T日订单结算与工作快照" in body
+    assert "字段：交收台账\n期望：交收推演" in body
+    assert "关联报表\n字段：发行 OTC003\n期望：名义本金 = 执行名义" in body
+
+
 def test_dimension_report_body_keeps_correct_rows_on_error_theme() -> None:
     """报表卡即使总结论是错误，正确列仍留在分报表，错误行逐条写期望和实据。"""
 

code/post_clear_check/tests/test_orchestrate.py

@@ -406,7 +406,10 @@ def test_cards_group_into_four_dimensions() -> None:
     assert "订单B" in order_formulas
     assert "发行A" not in order_formulas
     assert "销售带订单A" not in order_formulas
-    assert any(text.startswith("关联报表索引：OTC003") for text in order_formulas)
+    assert any(text.startswith("关联报表索引：发行 OTC003") for text in order_formulas)
+    assert all(len(str(row["report_code"])) <= 16 for row in conn.lines)
+    assert any(text.startswith("清算对照：交收台账") for text in order_formulas)
+    assert any(text.startswith("清算对照：销售资金流水") for text in order_formulas)
     assert any("发行 OTC003：本轮主单 2 笔，正确 2" in row["calc_text"] for row in conn.lines)
     assert "别的编号" not in order_formulas
     assert _formulas(conn, day, DIM_ISSUE) == ["发行A", "发行B", "别的编号"]

doc/使用说明.md

@@ -176,7 +176,7 @@ OTC009、OTC012 是实时导出，没有落库结果，`report_topup_payment`、
 
 飞书按订单链路、发行报表、销售报表、TRS 报表的顺序发。标题形如 `订单链路 可疑`、`发行报表 正确`。失败重发在末尾加「重发」。标题不写系统和清算步骤。颜色按这张卡的总结论：有错误为红，否则有可疑为橙，否则为蓝。
 
-订单链路正文是摘要、每单链路、盘中公式、清算对照、关联报表、错误清单、可疑清单、无法展开。摘要里的笔数按主单计。每单链路写两段：执行事实是否与实时源表一致；需要 TRS 执行根的对冲，成交股数是否对上，并说明自动单按 ext_ref、补录按委托号或母单号。不列表，也不把未核对的票据和已撤单写进去。同一字段的正确行收成一块，结论写「正确（n/n）」。关联报表只写本轮主单在三张报表卡上的对错笔数，不把报表列再算一遍。三端门禁在摘要里各一行。
+订单链路正文是摘要、每单链路、盘中公式、清算对照、关联报表、错误清单、可疑清单、无法展开。摘要里的笔数按主单计。每单链路写两段：执行事实是否与实时源表一致；需要 TRS 执行根的对冲，成交股数是否对上，并说明自动单按 ext_ref、补录按委托号或母单号。不列表，也不把未核对的票据和已撤单写进去。同一字段的正确行收成一块，结论写「正确（n/n）」。清算对照除订单自己的已发布执行列、工作快照外，再写交收台账、销售资金流水、TRS 发生额与汇率、清算留痕；没有对应结论行就写「本轮未核」，不编造金额。关联报表按发行/销售的 OTC003、OTC004、OTC007/OTC008 展开期望、实据和依据。三端门禁在摘要里各一行。
 
 三张报表正文是批次门禁、摘要、分报表、错误清单、可疑清单、无法展开。没有门禁行时不写批次门禁。同一报表、同一依据的正确列收成一块。错误和可疑逐条写节点、期望、实据、依据、结论。方向卡是错误主题时，正确列仍留在分报表。