fix(digest): Bind IBKR account display to the Telegram recipient

docs/daily-digest-display-and-resend.md

@@ -35,6 +35,32 @@ existing notification mode and private candidate configuration remain in use.
 
 ## Account display
 
+Use `python/scripts/project_digest_display_roster.py` when enrolling an
+account or reconciling names after a management configuration edit. Give it
+the current protected digest roster, the authenticated management
+runtime-instances response, and the private enrollment links. Each link
+contains the digest's `platform_id`, `candidate_slot`, `opaque_account_uid`,
+and `target_id`, the management `account_key`, and the verified management
+`config_identity` fields `account_selector`, `deployment_selector`,
+`service_name`, and `runtime_status_target_id`. Keep those inputs outside Git.
+
+`--check` reports display drift. `--output` prepares a private replacement
+roster for the existing release procedure. The projection changes account
+labels and row order only. It removes IBKR roster aliases. The authorized
+Telegram account field comes from the verified broker selector, independently
+of any strategy or runtime label change. A changed binding requires verifying
+the enrollment link before applying a display update.
+
+The central Azure scheduled sender can additionally show full IBKR account IDs in the
+authorized Telegram recipient's message. Set the private
+`DIGEST_TELEGRAM_IBKR_ACCOUNT_IDS_JSON` to the approved `recipient_id` and an
+`accounts` array, binding each `account_id` to its exact `candidate_slot`,
+`opaque_account_uid`, and `target_id`. The sender falls back to the existing
+safe account display if the recipient or a binding changes. Previews, receipts and
+other notification paths keep their existing safe account displays.
+For an approved release, `--output` with `--telegram-recipient-file` emits both
+private settings together from the same verified roster and configuration.
+
 When **both** fill and order counts are known and zero, the account shows
 `无交易` / `No trades`. Accounts with activity retain their fill/order counts.
 A missing count gets an explicit reason instead of an invented zero. Any

python/scripts/account_daily_digest.py

@@ -104,7 +104,8 @@ def _snapshot_label(stamp, window, zh):
     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):
+def render_account_daily(payload, expected, outcomes, *, facts=None, known_platforms=(), max_units=3850,
+                         telegram_ibkr_accounts=None):
     """Render supplied raw binding outcomes and optional per-account evidence.
 
     ``facts`` is keyed by (platform_id, opaque_account_uid). Order and fill IDs
@@ -117,6 +118,7 @@ def render_account_daily(payload, expected, outcomes, *, facts=None, known_platf
     unknown = '未知' if zh else 'unknown'
     window = sgt_business_window(datetime.combine(date.fromisoformat(payload.business_day), time(6), SGT))
     facts = facts or {}
+    telegram_ibkr_accounts = telegram_ibkr_accounts or {}
     groups = {}
     for i, binding in enumerate(expected):
         key = (binding.get('platform_id', ''), binding.get('opaque_account_uid', ''))
@@ -150,6 +152,8 @@ def render_account_daily(payload, expected, outcomes, *, facts=None, known_platf
         'invalid': ('报告无效', 'Report invalid'),
     }
     for ((platform, uid), group), (platform_label, account, _) in zip(groups.items(), contexts):
+        if platform == 'ibkr':
+            account = telegram_ibkr_accounts.get(uid, account)
         rows, bindings = group['rows'], group['bindings']
         actual_rows = [row for row in rows if row.execution_mode.strip().lower() not in SIMULATION_MODES]
         has_simulation = len(actual_rows) != len(rows)

python/scripts/cloud_run_daily_digest.py

@@ -25,12 +25,13 @@
 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
+from display_names import broker_account_identity, 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'
+IBKR_TELEGRAM_BINDING = ('candidate_slot', 'opaque_account_uid', 'target_id')
 
 
 class _NoRedirect(urllib.request.HTTPRedirectHandler):
@@ -52,6 +53,29 @@ def project_expected_target_label(binding, account_option):
     return result
 
 
+def _ibkr_telegram_accounts(expected, raw, recipient):
+    if not raw or not recipient:
+        return {}
+    try:
+        value = json.loads(raw)
+        if value['recipient_id'] != recipient:
+            return {}
+        eligible = {tuple(row[key] for key in IBKR_TELEGRAM_BINDING)
+                    for row in expected if row['platform_id'] == 'ibkr'}
+        accounts, owners = {}, {}
+        for row in value['accounts']:
+            identity = tuple(row[key] for key in IBKR_TELEGRAM_BINDING)
+            uid = row['opaque_account_uid']
+            account = broker_account_identity('ibkr', row['account_id'])
+            if (identity not in eligible or not account
+                    or accounts.setdefault(uid, account) != account
+                    or owners.setdefault(account, uid) != uid):
+                return {}
+        return accounts
+    except (KeyError, TypeError, ValueError):
+        return {}
+
+
 @dataclass(frozen=True)
 class Configuration:
     expected: list
@@ -63,6 +87,7 @@ class Configuration:
     candidate_uris: dict = field(default_factory=dict)
     binance_snapshot_endpoint: str = ''
     binance_snapshot_token: str = field(default='', repr=False)
+    telegram_ibkr_accounts: dict = field(default_factory=dict, repr=False)
 
     def __post_init__(self):
         adapter.validate_expected(self.expected)
@@ -129,12 +154,15 @@ def from_environment(cls):
             candidate_uris = json.loads(os.environ.get('DIGEST_CANDIDATES_BY_PLATFORM_JSON', '{}'))
         except ValueError:
             raise ValueError('digest_candidate_sources_invalid') from None
-        return cls(json.loads(os.environ['DIGEST_EXPECTED_TARGETS_JSON']),
+        expected = json.loads(os.environ['DIGEST_EXPECTED_TARGETS_JSON'])
+        return cls(expected,
                    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('DIGEST_BINANCE_SNAPSHOT_ENDPOINT', ''),
-                   os.environ.get('DIGEST_BINANCE_SNAPSHOT_TOKEN', ''))
+                   os.environ.get('DIGEST_BINANCE_SNAPSHOT_TOKEN', ''),
+                   _ibkr_telegram_accounts(expected, os.environ.get('DIGEST_TELEGRAM_IBKR_ACCOUNT_IDS_JSON'),
+                                           os.environ.get('QSL_GLOBAL_TELEGRAM_CHAT_ID')))
 
 
 def utc_now():
@@ -289,7 +317,8 @@ def _ibkr_settled_cash(raw, binding, day):
     return _SettledCashObservation(binding['opaque_account_uid'], raw['amount'], statement_day)
 
 
-def _report(configuration, rows, outcomes, day, platform_config, *, stale_observations=()):
+def _report(configuration, rows, outcomes, day, platform_config, *, stale_observations=(),
+            include_telegram_account_ids=False):
     config = deepcopy(platform_config or aggregator.load_platform_config())
     config['notifications']['quant_sentinel']['daily_digest'].setdefault('aggregator', {})['expected_targets'] = configuration.expected
     failure_reasons = {'read_failed': 'candidate_store_read_failed',
@@ -315,7 +344,9 @@ def _report(configuration, rows, outcomes, day, platform_config, *, stale_observ
     if coverage == 'complete' and (payload.notification_summary or {}).get('counts_unknown'):
         coverage = 'partial'
     platforms = dict.fromkeys((*adapter.PLATFORMS, *config.get('platforms', {})))
-    text = render_account_daily(payload, configuration.expected, outcomes, facts=facts, known_platforms=platforms)
+    text = render_account_daily(payload, configuration.expected, outcomes, facts=facts, known_platforms=platforms,
+                                telegram_ibkr_accounts=(configuration.telegram_ibkr_accounts
+                                                        if include_telegram_account_ids else None))
     return text, coverage
 
 
@@ -367,7 +398,9 @@ def execute_slot(report_date, day, check, schedule_record, configuration, store,
         read_end = now() + timedelta(seconds=120) if preview else stop
         rows, outcomes, stale = _sources(
             configuration, store, day, read_end, now=now, monotonic=monotonic)
-        text, coverage = _report(configuration, rows, outcomes, day, platform_config, stale_observations=stale)
+        text, coverage = _report(configuration, rows, outcomes, day, platform_config, stale_observations=stale,
+                                 include_telegram_account_ids=(
+                                     not preview and schedule_record.get('trigger_provenance') == 'azure_logic_sliding_window'))
     # Recheck after intake/rendering: wall-clock expiry cannot be extended by retries.
     late = late or now() + timedelta(seconds=30) >= deadline_at
     if late and not preview:

python/scripts/project_digest_display_roster.py

@@ -0,0 +1,136 @@
+"""Prepare a private, display-only digest roster from the current account configuration."""
+from __future__ import annotations
+
+import argparse
+import json
+import os
+from pathlib import Path
+
+from display_names import broker_account_identity, safe_display_text
+
+
+# The digest target can differ from the management instance key. Compare the
+# private enrollment link against the actual digest binding before using it.
+BINDING_FIELDS = ('platform_id', 'candidate_slot', 'opaque_account_uid', 'target_id')
+CONFIG_IDENTITY_FIELDS = ('account_selector', 'deployment_selector', 'service_name', 'runtime_status_target_id')
+
+
+def _binding_key(row):
+    key = tuple(row.get(field) for field in BINDING_FIELDS)
+    if not all(isinstance(value, str) and value for value in key):
+        raise ValueError('invalid_digest_binding')
+    return key
+
+
+def _display_label(platform, config, binding):
+    label = config.get('label')
+    if not isinstance(label, str) or safe_display_text(label) != label or not label:
+        raise ValueError('invalid_configured_account_label')
+    # The two channel names are display text even when the private LongBridge
+    # target uses the same lowercase abbreviation.
+    if platform == 'longbridge' and label in {'hk', 'sg'}:
+        return label.upper()
+    if label in {binding['opaque_account_uid'], binding['target_id'], binding.get('account_hint')}:
+        raise ValueError('configured_label_is_private_identity')
+    return label
+
+
+def _project(roster, management, enrollment_links):
+    if management.get('initialized') is not True:
+        raise ValueError('account_configuration_unavailable')
+    configured, platform_order = {}, {}
+    for index, instance in enumerate(management['instances']):
+        if instance.get('kind') != 'existing':
+            continue
+        config = instance['config']
+        platform = instance['platform']
+        platform_order.setdefault(platform, len(platform_order))
+        key = platform, config['key']
+        if key in configured:
+            raise ValueError('duplicate_configured_account')
+        configured[key] = index, config
+
+    links = {_binding_key(row): row for row in enrollment_links}
+    bindings = {_binding_key(row) for row in roster}
+    if len(links) != len(enrollment_links) or len(bindings) != len(roster) or links.keys() != bindings:
+        raise ValueError('digest_enrollment_links_do_not_match_roster')
+    owners = {}
+    ranked, accounts = [], []
+    for binding in roster:
+        identity = _binding_key(binding)
+        platform, _, account_uid, _ = identity
+        link = links[identity]
+        key = platform, link['account_key']
+        if key not in configured:
+            raise ValueError('digest_configured_account_missing')
+        index, config = configured[key]
+        config_identity = {field: config.get(field) for field in CONFIG_IDENTITY_FIELDS}
+        if config_identity != link.get('config_identity') or not config_identity['account_selector']:
+            raise ValueError('digest_configured_account_identity_changed')
+        if owners.setdefault(key, account_uid) != account_uid:
+            raise ValueError('digest_configured_account_link_ambiguous')
+        projected = dict(binding)
+        if platform == 'ibkr':
+            account = broker_account_identity(platform, config['account_selector'])
+            if not account:
+                raise ValueError('invalid_ibkr_account_selector')
+            projected.pop('account_label', None)
+            accounts.append({**{field: binding[field] for field in BINDING_FIELDS if field != 'platform_id'},
+                             'account_id': account})
+        else:
+            projected['account_label'] = _display_label(platform, config, binding)
+        ranked.append(((platform_order[platform], index), projected))
+
+    return [binding for _, binding in sorted(ranked, key=lambda row: row[0])], accounts
+
+
+def project_display_roster(roster, management, enrollment_links):
+    return _project(roster, management, enrollment_links)[0]
+
+
+def project_telegram_display_settings(roster, management, enrollment_links, recipient):
+    if not isinstance(recipient, str) or not recipient.strip():
+        raise ValueError('telegram_recipient_required')
+    projected, accounts = _project(roster, management, enrollment_links)
+    return {'DIGEST_EXPECTED_TARGETS_JSON': json.dumps(projected, ensure_ascii=False),
+            'DIGEST_TELEGRAM_IBKR_ACCOUNT_IDS_JSON': json.dumps({'recipient_id': recipient, 'accounts': accounts})}
+
+
+def main():
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument('--roster', type=Path, required=True)
+    parser.add_argument('--management', type=Path, required=True)
+    parser.add_argument('--enrollment-links', type=Path, required=True)
+    output = parser.add_mutually_exclusive_group(required=True)
+    output.add_argument('--check', action='store_true')
+    output.add_argument('--output', type=Path)
+    parser.add_argument('--telegram-recipient-file', type=Path)
+    args = parser.parse_args()
+    if args.telegram_recipient_file and args.check:
+        parser.error('--telegram-recipient-file requires --output')
+    try:
+        roster, management, links = (
+            json.loads(path.read_text(encoding='utf-8'))
+            for path in (args.roster, args.management, args.enrollment_links)
+        )
+        if args.telegram_recipient_file:
+            projected = project_telegram_display_settings(
+                roster, management, links, args.telegram_recipient_file.read_text(encoding='utf-8').strip(),
+            )
+        else:
+            projected = project_display_roster(roster, management, links)
+        if args.check:
+            print('digest_display_current' if roster == projected else 'digest_display_out_of_sync')
+            return 0 if roster == projected else 1
+        descriptor = os.open(args.output, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600)
+        with os.fdopen(descriptor, 'w', encoding='utf-8') as handle:
+            json.dump(projected, handle, ensure_ascii=False, indent=2)
+            handle.write('\n')
+    except (AttributeError, KeyError, OSError, TypeError, ValueError):
+        print('digest_display_projection_failed')
+        return 2
+    return 0
+
+
+if __name__ == '__main__':
+    raise SystemExit(main())

python/tests/test_digest_config_labels.py

@@ -10,14 +10,60 @@
 
 import cloud_run_daily_digest as app
 import daily_digest_aggregator as aggregator
-from test_cloud_run_daily_digest import EXPECTED, JOB
+from test_cloud_run_daily_digest import EXPECTED, JOB, NOW, MemoryStore
 from test_notification_consolidation import candidate, build
 from test_unified_daily_digest_presentation import payload, row
 from unified_daily_digest import render
 from project_public_digest_receipt import project_public_receipt
 
 
 class ConfigLabelTests(unittest.TestCase):
+    def test_full_ibkr_account_is_only_rendered_for_bound_azure_delivery(self):
+        binding = {**EXPECTED[0], 'platform_id': 'ibkr'}
+        account = {**{key: binding[key] for key in app.IBKR_TELEGRAM_BINDING}, 'account_id': 'U0000099'}
+        env = {
+            'QRS_UNIFIED_NOTIFICATION_ENABLED': 'true',
+            'DIGEST_DELIVERY_ENABLED': 'true',
+            'DIGEST_EXPECTED_TARGETS_JSON': json.dumps([binding]),
+            'DIGEST_SCHEDULER_JOB_NAME': JOB,
+            'DIGEST_RECIPIENT_SCOPE': 'synthetic',
+            'DIGEST_CUTOVER_REPORT_DATE': '2026-10-16',
+            'QSL_GLOBAL_TELEGRAM_CHAT_ID': 'test-recipient',
+            'DIGEST_TELEGRAM_IBKR_ACCOUNT_IDS_JSON': json.dumps({
+                'recipient_id': 'test-recipient', 'accounts': [account],
+            }),
+        }
+        with patch.dict(os.environ, env, clear=True):
+            configured = app.Configuration.from_environment()
+        self.assertNotIn(account['account_id'], repr(configured))
+        self.assertNotIn(account['account_id'], json.dumps(configured.expected))
+        text, _ = app._report(configured, [], ['missing'], '2026-10-15', None)
+        self.assertIn('IBKR · 账户', text)
+        self.assertNotIn(account['account_id'], text)
+        for provenance, displays_account in (('azure_logic_sliding_window', True),
+                                             ('azure_functions_timer', False), ('cloud_scheduler', False)):
+            with self.subTest(provenance=provenance):
+                store, sent = MemoryStore({}), []
+
+                def send(body):
+                    sent.append(body)
+                    return 1
+
+                schedule = {'scheduled_at': NOW.isoformat(), 'trigger_provenance': provenance}
+                args = (NOW.date(), '2026-10-15', False, schedule, configured, store, send)
+                _, preview = app.execute_slot(*args, preview=True, now=lambda: NOW)
+                self.assertEqual(preview['delivery'], 'preview')
+                self.assertFalse(sent)
+                _, result = app.execute_slot(*args, now=lambda: NOW)
+                self.assertEqual(result['delivery'], 'accepted')
+                self.assertEqual(account['account_id'] in sent[0], displays_account)
+                self.assertNotIn(account['account_id'], json.dumps(result))
+                self.assertNotIn(account['account_id'], json.dumps(store.records))
+        for changed_account, recipient in ((account, 'different-recipient'),
+                                           ({**account, 'target_id': 'other-target'}, 'test-recipient')):
+            data = json.dumps({'recipient_id': 'test-recipient', 'accounts': [changed_account]})
+            self.assertEqual(app._ibkr_telegram_accounts([binding], data, recipient), {})
+
     def test_projection_precedence_and_no_mutation(self):
         for option, expected in (({'label': 'Custom', 'key': 'live', 'target_name': 'sg'}, 'Custom'),
                                  ({'label': '', 'key': 'paper', 'target_name': 'sg'}, 'paper'),