fix: Preserve verified daily account observations

docs/unified-daily-notification.md

@@ -93,12 +93,17 @@ remain valid; its presence alone does not establish account-wide coverage.
 The first release reads the existing bound candidates only. It reports timed
 fills found in those candidates. A fill stored solely under another business
 day or in a separate supplemental object is outside the observed scope.
-Assets and positions come from one bound observation. An earlier snapshot is
-dated, legacy observations without a snapshot time are labelled, and partial
-positions are labelled as such. Missing assets or positions
-do not suppress the other verified facts from the same observation. Unavailable positions and
-confirmed empty positions remain distinct. The single-message budget preserves
-account summaries and states how many positions are shown and omitted.
+Assets and positions each use the most recent verified observation for the
+bound account. If the timestamps differ, both are shown. Earlier snapshots,
+legacy observations without a snapshot time, and partial positions are labelled.
+Unavailable positions and confirmed empty positions remain distinct. The
+single-message budget preserves account summaries and states how many positions
+are shown and omitted.
+
+When run counts may overlap, the report shows the largest verified count or
+number of distinct observed records, whichever is higher. The total remains
+unknown. Approved risk blocks show the verified reason. An unconfirmed execution
+only carries a pending order status when there is evidence of an order.
 
 The account report omits runtime links and technical evidence. The compact
 deadline fallback keeps trades, assets and holdings unknown and retains the

python/scripts/account_daily_digest.py

@@ -10,6 +10,7 @@
 from account_daily_facts import SIMULATION_MODES
 from daily_digest_notify import normalize_locale
 from display_names import broker_account_identity, display_name, safe_display_text
+from notification_approved_gates import render_reasons
 from notification_snapshot.unified_contract import SGT, sgt_business_window
 from unified_daily_digest import (
     _account_identity_label,
@@ -87,6 +88,14 @@ def _count_value(count, observed, zh):
     return '未知' if zh else 'unknown'
 
 
+def _snapshot_label(stamp, window, zh):
+    if stamp < window.start:
+        prefix = '较早快照 ' if zh else 'Earlier snapshot '
+    else:
+        prefix = '快照 ' if zh else 'Snapshot '
+    return prefix + stamp.astimezone(SGT).strftime('%Y-%m-%d %H:%M UTC+8')
+
+
 def render_account_daily(payload, expected, outcomes, *, facts=None, known_platforms=(), max_units=3850):
     """Render supplied raw binding outcomes and optional per-account evidence.
 
@@ -179,13 +188,29 @@ def render_account_daily(payload, expected, outcomes, *, facts=None, known_platf
             else:
                 count = None
             counts[name] = count
-            value = _count_value(count, len(records), zh)
+            observed = len(records)
+            if count is None and not (name == 'fill' and legacy_fill_window_unreliable):
+                # Counts from different runs may cover the same broker records.
+                observed = max([observed, *(getattr(row, name + '_count') for row in rows if _known_count(row, name))])
+            value = _count_value(count, observed, zh)
             count_lines.append(f'{label} {value}' if zh else f'{label}: {value}')
-        if any(_execution_blocked(row) for row in rows):
+        blocked = [row for row in rows if _execution_blocked(row)]
+        if any(not row.approved_gate_reasons for row in blocked):
             details.append('执行已拦截' if zh else 'Execution blocked')
-        if any(row.status == 'alert' for row in rows):
+        details.extend(dict.fromkeys(render_reasons(row.approved_gate_reasons, locale)
+                                    for row in blocked if row.approved_gate_reasons))
+        if any(row.status == 'alert' and not row.approved_gate_reasons and not _execution_unconfirmed(row)
+               for row in rows):
             details.append('执行异常' if zh else 'Execution issue')
-        if any(_execution_unconfirmed(row) for row in rows):
+        unconfirmed = [row for row in rows if _execution_unconfirmed(row)]
+        order_states = {'order_submitted', 'submitted', 'broker_acknowledged', 'partially_filled'}
+        def has_order_evidence(row):
+            return (row.execution_status.strip().lower() in order_states
+                    or (_known_count(row, 'order') and row.order_count > 0))
+
+        if any(not has_order_evidence(row) for row in unconfirmed):
+            details.append('执行结果待核对' if zh else 'Execution result unconfirmed')
+        if any(has_order_evidence(row) for row in unconfirmed):
             details.append('订单最终状态待确认' if zh else 'Final order state unconfirmed')
         details.extend(dict.fromkeys(
             _evidence_attention_text(row, locale) for row in rows
@@ -236,14 +261,24 @@ def strategy_for(record):
                     snapshots.append((stamp, item))
             except ValueError:
                 pass
-        complete, holdings, equity, stamp_label = False, [], unknown, ''
+        complete, holdings, equity = False, [], unknown
+        stamp_label, holding_stamp_label = '', ''
         if snapshots:
-            stamp, snapshot = max(snapshots, key=lambda pair: pair[0])
-            stamp_label = ('较早快照 ' if zh else 'Earlier snapshot ') if stamp < window.start else ('快照 ' if zh else 'Snapshot ')
-            stamp_label += stamp.astimezone(SGT).strftime('%Y-%m-%d %H:%M UTC+8')
-            equity = _money(snapshot.get('equity'), snapshot.get('equity_currency'), unknown)
-            holdings = snapshot.get('holdings') or []
-            complete = snapshot.get('holdings_complete') is True
+            snapshots.sort(key=lambda pair: pair[0], reverse=True)
+            asset_stamp = None
+            for stamp, snapshot in snapshots:
+                equity = _money(snapshot.get('equity'), snapshot.get('equity_currency'), unknown)
+                if equity != unknown:
+                    asset_stamp = stamp
+                    stamp_label = _snapshot_label(stamp, window, zh)
+                    break
+            for stamp, snapshot in snapshots:
+                holdings = snapshot.get('holdings') or []
+                complete = snapshot.get('holdings_complete') is True
+                if holdings or complete:
+                    if stamp != asset_stamp:
+                        holding_stamp_label = _snapshot_label(stamp, window, zh)
+                    break
         else:
             observations = {(row.equity, row.equity_currency, row.holdings, row.holdings_scope)
                             for row in rows if row.equity is not None or row.holdings}
@@ -265,6 +300,8 @@ def strategy_for(record):
             holding_label += ('空仓' if zh else 'No holdings') if complete else unknown
         elif not complete:
             holding_label += '仅列已核实持仓' if zh else 'Verified positions only'
+        if holding_stamp_label:
+            holding_label += (' · ' if not holdings or not complete else '') + holding_stamp_label
         holding_lines = [
             f"{_symbol(h.get('symbol'), unknown)} {_quantity(h.get('quantity'), h.get('asset_type'), zh, unknown)}"
             for h in holdings

python/tests/test_daily_caller_presentation.py

@@ -95,6 +95,42 @@ def test_pending_partial_cancel_reject_and_expiry_follow_evidence(self):
                 if status in {"cancelled", "rejected", "expired"}:
                     self.assertNotIn("成交待确认", text)
 
+    def test_unconfirmed_execution_does_not_imply_an_order(self):
+        item = with_facts({**ROW, "status": "alert", "execution_status": "execution_unknown",
+                           "order_count": None, "fill_count": None})
+        text, _ = self.report([item], ["present"])
+        self.assertIn("执行结果待核对", text)
+        self.assertNotIn("订单最终状态待确认", text)
+
+        submitted = {**item, "execution_status": "order_submitted", "order_count": 1}
+        text, _ = self.report([submitted], ["present"])
+        self.assertIn("订单最终状态待确认", text)
+
+    def test_verified_risk_block_reports_the_reason_without_an_execution_error(self):
+        reason = {"reason_code": "consecutive_loss_limit_exceeded", "observed_value": 6, "limit": 5}
+        item = with_facts({**ROW, "status": "alert", "execution_status": "strategy_risk_rejected",
+                           "block_reason": reason, "approved_gate_reasons": [reason]})
+        text, _ = self.report([item], ["present"])
+        self.assertIn("连亏熔断", text)
+        self.assertIn("本轮未下单", text)
+        self.assertNotIn("执行异常", text)
+        self.assertNotIn("无交易", text)
+
+    def test_multiple_runs_keep_verified_count_as_a_minimum(self):
+        first = with_facts({**ROW, "order_count": 3, "fill_count": 2})
+        second = with_facts({**ROW, "source_run_id": "other-run", "order_count": 2, "fill_count": 1})
+        text, _ = self.report([first, second], ["present"])
+        self.assertIn("提交订单 已见 3 笔，总数未知", text)
+        self.assertIn("成交回报 已见 2 笔，总数未知", text)
+
+    def test_newer_partial_snapshot_preserves_other_verified_field_and_its_time(self):
+        before = snapshot(as_of="2026-10-16T04:00:00+08:00", holdings=None)
+        after = snapshot(equity=None, equity_currency=None)
+        text, _ = self.report([with_facts(ROW, snapshots=[after, before])], ["present"])
+        self.assertIn("总资产：USD 50,000.00 · 快照 2026-10-16 04:00", text)
+        self.assertIn("持仓：快照 2026-10-16 05:10", text)
+        self.assertIn("TQQQ 300股", text)
+
     def test_no_trade_keeps_assets_and_explicit_empty_positions(self):
         item = with_facts(ROW, snapshots=[snapshot(holdings=[])])
         text, coverage = self.report([item], ["present"])