feat: unify daily digest sections and source coverage

docs/unified-daily-notification.md

@@ -51,6 +51,39 @@ normalizing them before rendering; unsupported or malformed locales fail
 configuration validation. The default is Chinese. Locale does not change the
 recipient/date delivery reservation or permit a second daily send.
 
+## Unified operator display
+
+The consolidated renderer uses the shared QPK five-section display contract:
+Result, Scope, Evidence, Coverage and Next step. Chinese and English render the
+same facts. The pure helper is vendored under `notification_snapshot` with its
+own source manifest. The manifest records the upstream revision and hash,
+the local account-context changes, and the current file hash. The existing
+event and routing snapshots retain their original source revisions.
+
+Financial observations remain inside their originating active account block.
+Required diagnostics retain that account's display header, including when
+multiple accounts share the same failure text. Budget selection keeps account
+summaries before optional observations and never retains an observation after
+omitting its account context.
+
+Every daily report lists IBKR, Schwab, Firstrade, LongBridge, Binance and the
+management console. An explicitly unbound source says it is not included and
+its run state is unknown. A missing inventory says coverage is unknown. This
+display does not bind a new source, collect another platform, or make unknown
+orders/fills zero. Existing totals cover bound targets only.
+
+Verified idle reports retain the business window, independently known counts,
+source coverage and next action. Their result says “No trades in verified
+targets”; unbound sources do not support an all-account claim. The one-message
+3900 UTF-16-unit budget reserves these sections and reports omitted evidence.
+An artificially small budget that cannot contain the mandatory fields fails
+with `notification_required_fields_exceed_budget` rather than dropping them.
+
+Existing receipt, delivery, routing, business-window and execution semantics
+are unchanged. QPK risk/guard/receipt templates adopt the same sections at
+source level; consumers need separate reviewed adoption. No source PR proves
+runtime adoption, a complete data connection or actual notification delivery.
+
 ## IBKR post-submit settlement
 
 The optional IBKR settlement feature starts with

python/scripts/notification_snapshot/presentation-source.json

@@ -0,0 +1,8 @@
+{
+  "repository": "QuantStrategyLab/QuantPlatformKit",
+  "commit": "fb8e687fb9a5d61694d5ad21591c52247bf8dfa1",
+  "path": "src/quant_platform_kit/notifications/presentation.py",
+  "sha256": "dca21d14215dfb9c18fe9798ca9f0fcb17a9482def69adb7df971c280501ad9f",
+  "upstream_sha256": "e3cfdadb0c138daa2e174ae1a94f43a81abacdd70b2c66b502443da4bc08e999",
+  "local_changes": "Preserve account context for required diagnostics and optional observations during budget selection."
+}

python/scripts/notification_snapshot/presentation.py

@@ -0,0 +1,102 @@
+"""Shared display sections; no execution, routing, receipt or transport policy."""
+from __future__ import annotations
+
+from collections.abc import Sequence
+from urllib.parse import urlsplit, urlunsplit
+
+SECTION_NAMES = {
+    "zh": ("结果", "范围", "证据", "覆盖", "下一步"),
+    "en": ("Result", "Scope", "Evidence", "Coverage", "Next step"),
+}
+
+
+def safe_run_url(value: object) -> str | None:
+    """Use only caller-supplied HTTPS links, dropping query and fragment."""
+    if not isinstance(value, str) or any(c.isspace() or ord(c) < 32 for c in value):
+        return None
+    try:
+        parsed = urlsplit(value)
+        if parsed.scheme == "https" and parsed.hostname and not parsed.username and not parsed.password:
+            return urlunsplit((parsed.scheme, parsed.netloc, parsed.path, "", ""))
+    except ValueError:
+        pass
+    return None
+
+
+def next_step_text(locale: str, *, needs_review: bool) -> str:
+    if locale == "zh":
+        return ("核对该次运行、回执和缺失来源；保持现有执行限制。" if needs_review else
+                "无需操作；等待下一统计窗口。")
+    return ("Review the source run, receipt and missing sources; retain existing execution restrictions."
+            if needs_review else "No action required; await the next reporting window.")
+
+
+def render_operator_message(*, title: str, locale: str, result: Sequence[str],
+                            scope: Sequence[str], evidence: Sequence[str], coverage: Sequence[str],
+                            next_step: Sequence[str], max_units: int | None = None,
+                            required_evidence: Sequence[str] = (),
+                            optional_evidence: Sequence[Sequence[str]] = ()) -> str:
+    """Keep result, scope, coverage and next step before optional evidence.
+
+    Truncation is explicit and cannot remove a section or imply full coverage.
+    Optional observations correspond to evidence entries in the same order.
+    Callers put urgent evidence before optional observations. Financial facts
+    must already have been validated by the producer; this helper infers none.
+    """
+    names = SECTION_NAMES["zh" if locale == "zh" else "en"]
+    unknown = "未知" if locale == "zh" else "Unknown"
+    groups = [[str(line) for line in group if line] or [unknown]
+              for group in (result, scope, evidence, coverage, next_step)]
+
+    def assemble():
+        lines = [title]
+        for name, group in zip(names, groups):
+            lines.extend([f"【{name}】" if locale == "zh" else f"[{name}]", *group])
+        return "\n".join(lines)
+
+    contexts = groups[2]
+
+    def observations(index):
+        return optional_evidence[index] if index < len(optional_evidence) else ()
+
+    groups[2] = [line for index, context in enumerate(contexts)
+                 for line in (context, *observations(index))]
+    if max_units is None:
+        return assemble()
+    if max_units < 256:
+        raise ValueError("notification_budget_too_small")
+    if len(assemble().encode("utf-16-le")) // 2 <= max_units:
+        return assemble()
+    original = contexts
+    # Retain caller order and duplicate reasons belonging to distinct accounts.
+    kept = {index for index, line in enumerate(original) if line in required_evidence}
+    suffix = "部分证据已省略；请查看运行记录。" if locale == "zh" else "Some evidence omitted; inspect source run."
+
+    def select_evidence():
+        groups[2] = [line for index, line in enumerate(original) if index in kept] + [suffix]
+
+    select_evidence()
+    if len(assemble().encode("utf-16-le")) // 2 > max_units:
+        raise ValueError("notification_required_fields_exceed_budget")
+    for index in range(len(original)):
+        if index in kept:
+            continue
+        kept.add(index)
+        select_evidence()
+        if len(assemble().encode("utf-16-le")) // 2 > max_units:
+            kept.remove(index)
+    # Add observations only after their retained account context, after all
+    # account summaries have had a chance to fit.
+    details = {}
+    for index, context in enumerate(original):
+        if index not in kept:
+            continue
+        for line in observations(index):
+            details.setdefault(index, []).append(line)
+            groups[2] = [item for i, entry in enumerate(original) if i in kept
+                         for item in (entry, *details.get(i, ()))] + [suffix]
+            if len(assemble().encode("utf-16-le")) // 2 > max_units:
+                details[index].pop()
+    groups[2] = [item for i, entry in enumerate(original) if i in kept
+                 for item in (entry, *details.get(i, ()))] + [suffix]
+    return assemble()

python/scripts/unified_daily_digest.py

@@ -6,6 +6,7 @@
 from datetime import date, datetime, time
 from notification_mode import consolidation_enabled, routine_mode
 from notification_snapshot.unified_contract import SGT, sgt_business_window
+from notification_snapshot.presentation import next_step_text, render_operator_message
 from display_names import broker_account_identity, display_name, runtime_account_names, safe_display_text as _safe_display_text
 from daily_digest_notify import DailyDigestInput, _presentation_needs_attention, _trade_counts, block_reason_text, counts_not_connected, is_blocked, normalize_locale, routine_decision_text, routine_observation_text
 
@@ -267,8 +268,7 @@ def _render_consolidated(payload: DailyDigestInput, *, max_units: int) -> str:
     summary = payload.notification_summary or {}
     coverage = payload.evidence_coverage
     complete = bool(coverage and coverage.is_complete_for_verified_idle and not summary.get("counts_scope_unknown", summary.get("counts_unknown")))
-    if complete and all(_verified_no_trade(row) for row in payload.runs):
-        return "😴 今日全部账号无交易" if zh else "😴 No trades across all accounts today"
+    verified_idle = complete and all(_verified_no_trade(row) for row in payload.runs)
     lines = ["📊 " + (f"每日交易摘要 · {payload.business_day}" if zh else f"Daily trading digest · {payload.business_day}")]
     lines.append("🕒 " + _window_text(payload.business_day, locale))
     lines.append(("窗口内运行次数未知" if zh else "Run count in window unknown") if summary.get("run_count_unknown") else
@@ -283,9 +283,15 @@ def _render_consolidated(payload: DailyDigestInput, *, max_units: int) -> str:
         _execution_unconfirmed(row) for row in payload.runs) else ""
     lines.append(" · ".join(counts) + confirmation)
     inventory = []
-    for platform, source in summary.get("source_inventory", {}).items():
+    sources = summary.get("source_inventory", {})
+    platforms = dict.fromkeys(("ibkr", "schwab", "firstrade", "longbridge", "binance", "qrs", *sources))
+    for platform in platforms:
+        source = sources.get(platform)
         fallback = ("管理站" if zh else "Management console") if platform == "qrs" else ""
         label = display_name("platforms", platform, locale, fallback=fallback)
+        if source is None:
+            inventory.append(f"{label}：来源覆盖未知" if zh else f"{label}: source coverage unknown")
+            continue
         expected, covered, runs = source["expected_targets"], source["covered_targets"], source["runs"]
         if not expected:
             inventory.append(f"{label}：未纳入本期汇总，运行状态未知" if zh else
@@ -306,7 +312,7 @@ def _render_consolidated(payload: DailyDigestInput, *, max_units: int) -> str:
         identity = row.opaque_account_uid or row.target_id or row.account_hint or ("row", index)
         accounts.setdefault((row.platform_id, identity), []).append((row, header, strategy))
     blocks, diagnostics, diagnostic_observations = [], [], []
-    required_diagnostics, compact_pending = set(), {}
+    required_diagnostics = []
     for entries in accounts.values():
         rows = [row for row, _, _ in entries]
         if all(_verified_no_trade(row) for row in rows):
@@ -352,59 +358,43 @@ def _render_consolidated(payload: DailyDigestInput, *, max_units: int) -> str:
         if needs_attention:
             block.append("需核对" if zh else "Needs review")
         block.extend(dict.fromkeys(pending))
-        required_diagnostics.update(details)
-        required_diagnostics.update(pending)
-        compact_pending.update((text, "订单最终状态待确认" if zh else "Final order state unconfirmed") for text in pending)
         details.extend(pending)
         optional = list(dict.fromkeys(text for row in rows for text in
                         (routine_decision_text(row, locale), routine_observation_text(row, locale)) if text))
         if any(_has_activity(row) for row in rows):
             optional.extend(_account_observations(rows, locale))
-            blocks.append((needs_attention, block, optional))
+            # Keep the header and required reasons atomic under truncation.
+            required = [block[0], *dict.fromkeys(details)] if details else []
+            blocks.append((needs_attention, block, optional, required))
         else:
             # Unknown counts are not trading activity. Keep safe diagnostics
             # without an account header, strategy block or account observations.
             diagnostics.extend(details)
             diagnostic_observations.extend(optional)
     if diagnostics or diagnostic_observations:
         blocks.append((True, ["需关注" if zh else "Diagnostics"] + list(dict.fromkeys(diagnostics)),
-                       list(dict.fromkeys(diagnostic_observations))))
+                       list(dict.fromkeys(diagnostic_observations)), list(dict.fromkeys(diagnostics))))
     blocks.sort(key=lambda item: not item[0])
-    footer = [] if complete else ["覆盖不完整，缺失统计保留未知" if zh else "Coverage incomplete; missing counts remain unknown"]
-    omitted = "部分观察字段已省略" if zh else "Some observations omitted"
-    # Reserve every account's counts and diagnostics before optional snapshots.
-    required = lines + [line for _, block, _ in blocks for line in block] + footer
-    potential_lines = len(required) + len(inventory) + sum(len(optional) for _, _, optional in blocks) + 1
-    suffix = f"其余 {potential_lines} 项已省略" if zh else f"{potential_lines} more items omitted"
-    budget = _units("\n".join(required)) + 1 + max(_units(omitted), _units(suffix))
-    observations_omitted = False
-    for _, block, optional in blocks:
-        if budget + _units("\n".join(optional)) + bool(optional) <= max_units:
-            block.extend(optional)
-            budget += _units("\n".join(optional)) + bool(optional)
-        elif optional:
-            observations_omitted = True
-        lines.extend(block)
-    lines.extend(footer)
-    if observations_omitted:
-        lines.append(omitted)
-    # Coverage inventory is supplementary and cannot displace trading accounts.
-    lines.extend(inventory)
-    # Compact finality before dropping context, and reserve every pending
-    # diagnostic before accepting dispensable window/count/account lines.
-    if _units("\n".join(lines)) > max_units:
-        lines = [compact_pending.get(line, line) for line in lines]
-        required_diagnostics = {compact_pending.get(line, line) for line in required_diagnostics}
-    pending_units = sum(_units(line) + 1 for line in lines[1:] if line in required_diagnostics)
-    result, skipped = [lines[0]], 0
-    suffix = f"其余 {len(lines)} 项已省略" if zh else f"{len(lines)} more items omitted"
-    for line in lines[1:]:
-        if line in required_diagnostics:
-            pending_units -= _units(line) + 1
-        if _units("\n".join(result + [line, suffix])) + pending_units > max_units:
-            skipped += 1
-            continue
-        result.append(line)
-    if skipped:
-        result.append(f"其余 {skipped} 项已省略" if zh else f"{skipped} more items omitted")
-    return "\n".join(result)
+    coverage_lines = ["统计范围：已绑定目标；缺失统计保留未知" if zh else
+                      "Counts cover bound targets; missing counts remain unknown"]
+    coverage_lines.extend(inventory)
+    evidence, optional_evidence = [], []
+    for _, block, optional, required in blocks:
+        context = "\n".join(required or block)
+        evidence.append(context)
+        optional_evidence.append(([line for line in block if line not in required]
+                                  if required else []) + optional)
+        if required:
+            required_diagnostics.append(context)
+    needs_review = (not complete or any(priority for priority, _, _, _ in blocks) or bool(immediate)
+                    or any(not _known_count(row, name) for row in payload.runs for name in ("order", "fill")))
+    result = [("证据需核对" if needs_review else "已收录窗口记录") if zh else
+              ("Evidence needs review" if needs_review else "Window observations recorded"), *lines[2:]]
+    if verified_idle:
+        result.insert(0, "😴 已核验目标无交易" if zh else "😴 No trades in verified targets")
+    return render_operator_message(
+        title=lines[0], locale=locale, result=result, scope=[lines[1]],
+        evidence=evidence, coverage=coverage_lines,
+        next_step=[next_step_text(locale, needs_review=needs_review)], max_units=max_units,
+        required_evidence=required_diagnostics, optional_evidence=optional_evidence,
+    )

python/tests/test_cloud_run_daily_digest.py

@@ -139,7 +139,7 @@ def test_sdk_reads_do_not_require_an_actions_candidate_status_variable(self):
         with patch.dict(os.environ, {'DIGEST_CANDIDATES_GCS_URI': 'gs://synthetic/candidates'}):
             _, result = self.run_request(configuration=self.enable())
         self.assertEqual(result['coverage'], 'complete')
-        self.assertEqual(self.sent[0], '😴 今日全部账号无交易')
+        self.assertIn('😴 已核验目标无交易', self.sent[0])
         self.assertNotIn('覆盖不完整', self.sent[0])
 
     def test_disabled_real_delivery_does_not_read_or_reserve(self):
@@ -221,7 +221,7 @@ def test_explicit_non_run_is_distinct_from_missing_without_inventing_its_cause(s
         self.store.rows[next(iter(self.store.rows))] = {'runs': [idle]}
         _, result = self.run_request(configuration=self.enable())
         self.assertEqual(result['coverage'], 'complete')
-        self.assertEqual(self.sent[0], '😴 今日全部账号无交易')
+        self.assertIn('😴 已核验目标无交易', self.sent[0])
         self.assertNotIn('休市', self.sent[0])
 
     def test_retry_keeps_original_report_date_and_rejects_another_day(self):

python/tests/test_daily_digest_v3.py

@@ -34,12 +34,12 @@ def test_active_accounts_are_flat_and_idle_account_is_omitted(self):
         for absent in ('00000003', '美股日策略', '定投', '加密小时策略', '提交订单', '单1/成1'):
             self.assertNotIn(absent, text)
 
-    def test_verified_idle_is_exactly_one_line_even_with_cycles_or_observations(self):
+    def test_verified_idle_keeps_scoped_summary_with_cycles_or_observations(self):
         item = row(order_count=0, fill_count=0, cycle_count=24, equity=1234, holdings=(DigestHolding('SYN', quantity=2),))
-        self.assertEqual(render(payload(item)), '😴 今日全部账号无交易')
-        self.assertEqual(render(payload()), '😴 今日全部账号无交易')
+        self.assertIn('😴 已核验目标无交易', render(payload(item)))
+        self.assertIn('😴 已核验目标无交易', render(payload()))
         blocked = render(payload(replace(item, status='alert', rebalance_conclusion='blocked')))
-        self.assertNotIn('今日全部账号无交易', blocked)
+        self.assertNotIn('已核验目标无交易', blocked)
         self.assertIn('已拦截', blocked)
 
     def test_only_provided_observation_fields_are_rendered_without_mutation(self):
@@ -66,10 +66,10 @@ def test_multiple_runs_of_one_account_have_one_header_and_summed_counts(self):
     def test_unknown_evidence_cannot_be_reported_as_idle(self):
         text = render(payload(row(order_count=1, fill_count=None, fill_count_status='unknown')))
         self.assertIn('提交1笔/成交未知笔', text)
-        self.assertNotIn('今日全部账号无交易', text)
+        self.assertNotIn('已核验目标无交易', text)
         from daily_digest_notify import EvidenceCoverage
         text = render(replace(payload(), evidence_coverage=EvidenceCoverage(candidate_failures=({'reason': 'read_failed'},))))
-        self.assertNotIn('今日全部账号无交易', text)
+        self.assertNotIn('已核验目标无交易', text)
         self.assertIn('未知', text)
 
     def test_holdings_are_sorted_capped_and_counted(self):
@@ -151,8 +151,8 @@ def test_accounts_without_known_activity_are_omitted_without_claiming_idle(self)
                         self.assertNotIn('00000003', text)
                         self.assertEqual(text.count('🏦 '), len(active))
                         self.assertIn('全窗口统计未知' if locale == 'zh' else 'whole-window count unknown', text)
-                        self.assertNotIn('今日全部账号无交易', text)
-                        self.assertNotIn('No trades across all accounts today', text)
+                        self.assertNotIn('已核验目标无交易', text)
+                        self.assertNotIn('No trades in verified targets', text)
 
     def test_active_account_retains_unknown_counts_from_another_run(self):
         active = row(order_count=1, fill_count=0)