feat: Read verified Binance holdings in the daily digest

docs/unified-daily-notification.md

@@ -105,6 +105,17 @@ 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.
 
+Binance can supply holdings from its existing Spot and Flexible Earn snapshot
+while trading is disabled. The console exposes the closed-window holdings for
+its bound account through a dedicated read-only credential. A scheduled digest
+binding opts in with `snapshot_only: true` and uses the console account key as
+its opaque account UID. The deployment supplies `DIGEST_BINANCE_SNAPSHOT_ENDPOINT`
+and `DIGEST_BINANCE_SNAPSHOT_TOKEN` through protected configuration. The reader
+checks the account, target and reporting window before displaying holdings.
+The archive retains the newest snapshot in each Singapore business window, so
+an observation after the cutoff cannot replace the prior window's snapshot.
+This snapshot leaves order counts, fills and account-wide valuation unknown.
+
 The account report omits runtime links and technical evidence. The compact
 deadline fallback keeps trades, assets and holdings unknown and retains the
 existing single reservation and no-follow-up behavior. Other notification

python/scripts/account_daily_digest.py

@@ -55,7 +55,15 @@ def _money(value, currency, unknown):
     return f'{currency} {text}'
 
 
-def _symbol(value, unknown):
+def valid_crypto_symbol(value):
+    return (isinstance(value, str) and 1 <= len(value) <= 128
+            and all(char.isalnum() and not ('a' <= char <= 'z') and char not in '\u115f\u1160\u3164\uffa0'
+                    for char in value))
+
+
+def _symbol(value, unknown, *, asset_type=None):
+    if asset_type == 'crypto' and valid_crypto_symbol(value):
+        return value
     value = str(value or '').strip()
     return value if re.fullmatch(r'[A-Za-z0-9][A-Za-z0-9.\-^/]{0,19}', value) else unknown
 
@@ -155,7 +163,7 @@ def render_account_daily(payload, expected, outcomes, *, facts=None, known_platf
         header = '🏦 ' + ' · '.join((platform_label, account, ' / '.join(strategies)))
         details = []
         for binding, outcome in bindings:
-            if outcome != 'present' and not (outcome == 'not_run' and len(bindings) == 1):
+            if outcome not in {'present', 'snapshot_only'} and not (outcome == 'not_run' and len(bindings) == 1):
                 fallback = '未命名策略' if zh else 'Unnamed strategy'
                 strategy = safe_display_text(display_name('strategies', binding.get('strategy_profile', ''), locale, fallback=fallback)) or fallback
                 text = outcome_text.get(outcome, ('状态未知', 'Status unknown'))[0 if zh else 1]
@@ -303,7 +311,7 @@ def strategy_for(record):
         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)}"
+            f"{_symbol(h.get('symbol'), unknown, asset_type=h.get('asset_type'))} {_quantity(h.get('quantity'), h.get('asset_type'), zh, unknown)}"
             for h in holdings
         ]
         # Every binding must have a matching run, not merely an outcome marker.
@@ -347,6 +355,8 @@ def strategy_for(record):
             state = '⏳ 已提交，成交待确认' if zh else '⏳ Submitted, fills unconfirmed'
         elif account_idle:
             state = '😴 无交易' if zh else '😴 No trades'
+        elif all(outcome == 'snapshot_only' for _, outcome in bindings):
+            state = '仅核实持仓快照，交易情况未知' if zh else 'Holdings snapshot verified, trading activity unknown'
         elif all(outcome == 'not_run' for _, outcome in bindings):
             state = '策略未运行，账户交易未知' if zh else 'Strategy did not run; account trades unknown'
         else:

python/scripts/cloud_run_daily_digest.py

@@ -1,8 +1,9 @@
-"""One authenticated scheduled report; no broker, GitHub, AI or control-plane calls."""
+"""Render one authenticated scheduled report from archived observations."""
 from __future__ import annotations
 
 import hashlib
 import json
+import math
 import os
 import re
 import tempfile
@@ -12,21 +13,29 @@
 from copy import deepcopy
 from dataclasses import dataclass, field
 from datetime import UTC, date, datetime, time, timedelta
+from http.client import HTTPException
 from pathlib import Path
+from urllib.error import HTTPError
 from uuid import uuid4
 from zoneinfo import ZoneInfo
 
 import daily_digest_aggregator as aggregator
 import merge_digest_candidates as candidate_contract
 import notification_candidate_adapter as adapter
-from account_daily_digest import render_account_daily, render_account_daily_unavailable
+from account_daily_digest import render_account_daily, render_account_daily_unavailable, valid_crypto_symbol
 from account_daily_facts import collect_account_facts
 from daily_digest_delivery_store import GCSDeliveryStore, StoreUnavailable, location
 from display_names import configured_account_label, safe_display_text as _safe_display_text
 
 SGT = ZoneInfo('Asia/Singapore')
 IDENTITY = adapter.IDENTITY[:4]
 SLOT = re.compile(r'[a-z0-9][a-z0-9_-]{0,40}')
+BINANCE_SNAPSHOT_PATH = '/api/internal/binance-account-facts/digest'
+
+
+class _NoRedirect(urllib.request.HTTPRedirectHandler):
+    def redirect_request(self, req, fp, code, msg, headers, newurl):
+        return None
 
 
 def project_expected_target_label(binding, account_option):
@@ -52,6 +61,8 @@ class Configuration:
     enabled: bool = False
     locale: str = 'zh'
     candidate_uris: dict = field(default_factory=dict)
+    binance_snapshot_endpoint: str = ''
+    binance_snapshot_token: str = field(default='', repr=False)
 
     def __post_init__(self):
         adapter.validate_expected(self.expected)
@@ -81,6 +92,7 @@ def __post_init__(self):
             except (TypeError, ValueError):
                 raise ValueError('digest_candidate_sources_invalid') from None
         identities, objects = set(), set()
+        snapshot_bindings = 0
         for row in self.expected:
             if not SLOT.fullmatch(row.get('candidate_slot', '')):
                 raise ValueError('digest_candidate_slot_required')
@@ -90,6 +102,12 @@ def __post_init__(self):
                 raise ValueError('digest_duplicate_target_binding')
             identities.add(identity)
             objects.add(object_key)
+            if 'snapshot_only' in row:
+                if row['platform_id'] != 'binance' or row['snapshot_only'] is not True:
+                    raise ValueError('digest_snapshot_binding_invalid')
+                snapshot_bindings += 1
+        if snapshot_bindings > 1:
+            raise ValueError('digest_snapshot_binding_invalid')
         date.fromisoformat(self.cutover_date)
         if not re.fullmatch(r'[a-zA-Z0-9_-]{1,64}', self.recipient_scope):
             raise ValueError('digest_recipient_scope_invalid')
@@ -114,7 +132,9 @@ def from_environment(cls):
         return cls(json.loads(os.environ['DIGEST_EXPECTED_TARGETS_JSON']),
                    os.environ['DIGEST_SCHEDULER_JOB_NAME'], os.environ['DIGEST_RECIPIENT_SCOPE'],
                    os.environ['DIGEST_CUTOVER_REPORT_DATE'], flag == 'true',
-                   os.environ.get('QSL_NOTIFY_LANG', 'zh'), candidate_uris)
+                   os.environ.get('QSL_NOTIFY_LANG', 'zh'), candidate_uris,
+                   os.environ.get('DIGEST_BINANCE_SNAPSHOT_ENDPOINT', ''),
+                   os.environ.get('DIGEST_BINANCE_SNAPSHOT_TOKEN', ''))
 
 
 def utc_now():
@@ -147,6 +167,52 @@ def report_slot(headers, configuration, now):
     return report_date, day, stamp, check
 
 
+def _binance_snapshot(configuration, binding, day, timeout):
+    endpoint = urllib.parse.urlsplit(configuration.binance_snapshot_endpoint)
+    if (endpoint.scheme != 'https' or not endpoint.hostname or endpoint.netloc != endpoint.hostname
+            or endpoint.path != BINANCE_SNAPSHOT_PATH or endpoint.query or endpoint.fragment
+            or not re.fullmatch(r'[A-Za-z0-9_-]{32,256}', configuration.binance_snapshot_token)):
+        raise StoreUnavailable()
+    query = urllib.parse.urlencode({'business_day': day})
+    request = urllib.request.Request(configuration.binance_snapshot_endpoint + '?' + query,
+                                     headers={'Authorization': 'Bearer ' + configuration.binance_snapshot_token,
+                                              'Accept': 'application/json'})
+    try:
+        with urllib.request.build_opener(_NoRedirect()).open(request, timeout=timeout) as response:
+            body = response.read(256 * 1024 + 1)
+    except HTTPError as error:
+        if error.code == 404:
+            return None
+        raise StoreUnavailable() from None
+    except (HTTPException, OSError):
+        raise StoreUnavailable() from None
+    if len(body) > 256 * 1024:
+        raise ValueError('digest_snapshot_invalid')
+    data = json.loads(body)
+    if (set(data) != {'ok', 'business_day', 'account_key', 'target_id', 'as_of', 'holdings'}
+            or data['ok'] is not True or data['business_day'] != day
+            or data['account_key'] != binding['opaque_account_uid'] or data['target_id'] != binding['target_id']):
+        raise ValueError('digest_snapshot_binding_mismatch')
+    observed = datetime.fromisoformat(data['as_of'])
+    start = datetime.combine(date.fromisoformat(day), time(6), SGT)
+    if observed.tzinfo is None or not start <= observed < start + timedelta(days=1):
+        raise ValueError('digest_snapshot_window_invalid')
+    if not isinstance(data['holdings'], list):
+        raise ValueError('digest_snapshot_invalid')
+    holdings = []
+    for item in data['holdings']:
+        symbol, quantity = item['asset'], item['quantity']
+        if not valid_crypto_symbol(symbol) or not isinstance(quantity, str):
+            raise ValueError('digest_snapshot_invalid')
+        amount = float(quantity)
+        if not math.isfinite(amount) or amount < 0:
+            raise ValueError('digest_snapshot_invalid')
+        if amount:
+            holdings.append({'symbol': symbol, 'quantity': amount, 'asset_type': 'crypto'})
+    return {**{name: binding[name] for name in IDENTITY}, 'business_day': day, 'actually_ran': False,
+            'account_daily_facts': {'snapshots': [{'as_of': data['as_of'], 'holdings': holdings}]}}
+
+
 def _sources(configuration, store, day, end, *, now, monotonic):
     rows, outcomes, stale_observations = [], [], []
     budget_end = monotonic() + min(120, max(0, (end - now()).total_seconds()))
@@ -156,6 +222,12 @@ def _sources(configuration, store, day, end, *, now, monotonic):
         if timeout > 0:
             key = f"{binding['platform_id']}/{day}/{binding['candidate_slot']}.json"
             try:
+                if binding.get('snapshot_only'):
+                    snapshot = _binance_snapshot(configuration, binding, day, timeout)
+                    if snapshot is not None:
+                        rows.append(snapshot)
+                    outcomes.append('snapshot_only' if snapshot is not None else 'missing')
+                    continue
                 raw = store.candidate(key, timeout)
                 if raw is None:
                     outcome = 'missing'
@@ -197,7 +269,7 @@ def _report(configuration, rows, outcomes, day, platform_config, *, stale_observ
             business_day=day, locale=configuration.locale, candidates_path=path,
             include_github=False, config=config, candidate_failures=failures, render_text=False,
         )
-    covered = sum(value in {'present', 'not_run'} for value in outcomes)
+    covered = sum(value in {'present', 'not_run', 'snapshot_only'} for value in outcomes)
     coverage = 'complete' if covered == len(outcomes) else 'partial' if covered else 'unavailable'
     facts = collect_account_facts([*rows, *stale_observations], configuration.expected, day)
     if coverage == 'complete' and (payload.notification_summary or {}).get('counts_unknown'):

python/tests/test_cloud_run_daily_digest.py

@@ -7,6 +7,7 @@
 import unittest
 from concurrent.futures import ThreadPoolExecutor
 from datetime import datetime, timedelta
+from http.client import IncompleteRead
 from pathlib import Path
 from unittest.mock import patch
 
@@ -87,6 +88,75 @@ def only_bound_candidate(key, timeout):
         self.assertIn('管理站汇总平台数据', text)
         self.assertEqual(text.count('🏦 '), 1)
 
+    def test_binance_snapshot_shows_holdings_and_keeps_trade_counts_unknown(self):
+        from dataclasses import replace
+
+        binding = {**EXPECTED[0], 'platform_id': 'binance', 'snapshot_only': True}
+        cfg = replace(self.enable(), expected=[binding],
+                      binance_snapshot_endpoint='https://console.example' + app.BINANCE_SNAPSHOT_PATH,
+                      binance_snapshot_token='synthetic-read-only-token-for-tests')
+        response = {'ok': True, 'business_day': '2026-10-15', 'as_of': '2026-10-15T16:00:00Z',
+                    'account_key': binding['opaque_account_uid'], 'target_id': binding['target_id'],
+                    'holdings': [{'asset': 'BTC', 'quantity': '0.0125'}, {'asset': '币', 'quantity': '1.5'},
+                                 {'asset': 'A' * 128, 'quantity': '3'}]}
+        with patch.object(app.urllib.request, 'build_opener') as opener:
+            opener.return_value.open.return_value = io.BytesIO(json.dumps(response).encode())
+            self.run_request(configuration=cfg, platform_config=config([binding]))
+            request = opener.return_value.open.call_args.args[0]
+            self.assertEqual(request.full_url, cfg.binance_snapshot_endpoint + '?business_day=2026-10-15')
+            self.assertIsNone(opener.call_args.args[0].redirect_request(None, None, 302, None, None, None))
+        text, = self.sent
+        self.assertIn('仅核实持仓快照，交易情况未知', text)
+        self.assertIn('BTC 0.0125', text)
+        self.assertIn('币 1.5', text)
+        self.assertIn('A' * 128 + ' 3', text)
+        self.assertIn('提交订单 未知｜成交回报 未知', text)
+        self.assertNotIn('策略未运行', text)
+
+    def test_binance_snapshot_rejects_wrong_account_or_out_of_window(self):
+        from dataclasses import replace
+
+        binding = {**EXPECTED[0], 'platform_id': 'binance', 'snapshot_only': True}
+        cfg = replace(self.configuration, expected=[binding],
+                      binance_snapshot_endpoint='https://console.example' + app.BINANCE_SNAPSHOT_PATH,
+                      binance_snapshot_token='synthetic-read-only-token-for-tests')
+        valid = {'ok': True, 'business_day': '2026-10-15', 'as_of': '2026-10-15T16:00:00Z',
+                 'account_key': binding['opaque_account_uid'], 'target_id': binding['target_id'],
+                 'holdings': []}
+        for fields in ({'account_key': 'wrong-account'}, {'as_of': '2026-10-15T22:00:00Z'}):
+            with self.subTest(fields=fields), patch.object(app.urllib.request, 'build_opener') as opener:
+                opener.return_value.open.return_value = io.BytesIO(json.dumps({**valid, **fields}).encode())
+                with self.assertRaises(ValueError):
+                    app._binance_snapshot(cfg, binding, '2026-10-15', 20)
+
+    def test_missing_binance_credentials_do_not_block_existing_accounts(self):
+        from dataclasses import replace
+
+        binding = {**EXPECTED[0], 'platform_id': 'binance', 'snapshot_only': True}
+        expected = [*EXPECTED, binding]
+        cfg = replace(self.enable(), expected=expected)
+        self.run_request(configuration=cfg, platform_config=config(expected))
+        text, = self.sent
+        self.assertIn('🏦 Schwab', text)
+        self.assertIn('🏦 Binance', text)
+        self.assertIn('报告读取失败', text)
+
+    def test_incomplete_binance_http_body_does_not_block_other_accounts(self):
+        from dataclasses import replace
+
+        binding = {**EXPECTED[0], 'platform_id': 'binance', 'snapshot_only': True}
+        expected = [*EXPECTED, binding]
+        cfg = replace(self.enable(), expected=expected,
+                      binance_snapshot_endpoint='https://console.example' + app.BINANCE_SNAPSHOT_PATH,
+                      binance_snapshot_token='synthetic-read-only-token-for-tests')
+        with patch.object(app.urllib.request, 'build_opener') as opener:
+            response = opener.return_value.open.return_value.__enter__.return_value
+            response.read.side_effect = IncompleteRead(b'{')
+            self.run_request(configuration=cfg, platform_config=config(expected))
+        text, = self.sent
+        self.assertIn('🏦 Schwab', text)
+        self.assertIn('报告读取失败', text)
+
     def test_filled_orders_count_distinct_orders_and_preserve_unknown_totals(self):
         key = next(iter(self.store.rows))
         fill = {'order_id': 'synthetic-order', 'occurred_at': '2026-10-15T22:00:00+08:00'}