From d5f0acb62cbb95ed9ee980ea1847ff49008efd59 Mon Sep 17 00:00:00 2001 From: Pigbibi <20649888+Pigbibi@users.noreply.github.com> Date: Thu, 12 Nov 2026 14:09:51 +0800 Subject: [PATCH] fix(digest): Bind IBKR account display to the Telegram recipient Co-Authored-By: Codex --- docs/daily-digest-display-and-resend.md | 26 ++++ python/scripts/account_daily_digest.py | 6 +- python/scripts/cloud_run_daily_digest.py | 45 +++++- .../scripts/project_digest_display_roster.py | 136 ++++++++++++++++++ python/tests/test_digest_config_labels.py | 48 ++++++- .../test_project_digest_display_roster.py | 94 ++++++++++++ 6 files changed, 347 insertions(+), 8 deletions(-) create mode 100644 python/scripts/project_digest_display_roster.py create mode 100644 python/tests/test_project_digest_display_roster.py diff --git a/docs/daily-digest-display-and-resend.md b/docs/daily-digest-display-and-resend.md index 215ab8a1..36375fa7 100644 --- a/docs/daily-digest-display-and-resend.md +++ b/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 diff --git a/python/scripts/account_daily_digest.py b/python/scripts/account_daily_digest.py index 253b50ac..7d55707e 100644 --- a/python/scripts/account_daily_digest.py +++ b/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) diff --git a/python/scripts/cloud_run_daily_digest.py b/python/scripts/cloud_run_daily_digest.py index 6790053e..73c5c7a3 100644 --- a/python/scripts/cloud_run_daily_digest.py +++ b/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: diff --git a/python/scripts/project_digest_display_roster.py b/python/scripts/project_digest_display_roster.py new file mode 100644 index 00000000..6fa082d6 --- /dev/null +++ b/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()) diff --git a/python/tests/test_digest_config_labels.py b/python/tests/test_digest_config_labels.py index 5a68452b..9ae58c54 100644 --- a/python/tests/test_digest_config_labels.py +++ b/python/tests/test_digest_config_labels.py @@ -10,7 +10,7 @@ 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 @@ -18,6 +18,52 @@ 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'), diff --git a/python/tests/test_project_digest_display_roster.py b/python/tests/test_project_digest_display_roster.py new file mode 100644 index 00000000..9e88c451 --- /dev/null +++ b/python/tests/test_project_digest_display_roster.py @@ -0,0 +1,94 @@ +import json +import unittest +from copy import deepcopy + +from project_digest_display_roster import ( + BINDING_FIELDS, + CONFIG_IDENTITY_FIELDS, + project_display_roster, + project_telegram_display_settings, +) + + +def binding(platform, slot, target): + return { + 'platform_id': platform, 'candidate_slot': slot, + 'opaque_account_uid': f'uid-{slot}', 'target_id': target, + 'strategy_profile': 'original_strategy', 'account_label': 'original label', + } + + +def instance(platform, key, label, selector): + return { + 'kind': 'existing', 'platform': platform, + 'config': { + 'key': key, 'label': label, 'account_selector': selector, + 'deployment_selector': key, 'service_name': f'{platform}-{key}', + 'runtime_status_target_id': f'{platform}.{key}', + }, + } + + +def link(row, item): + config = item['config'] + return { + **{field: row[field] for field in BINDING_FIELDS}, + 'account_key': config['key'], + 'config_identity': {field: config[field] for field in CONFIG_IDENTITY_FIELDS}, + } + + +class DisplayRosterTests(unittest.TestCase): + def setUp(self): + self.ibkr = instance('ibkr', 'legacy-key', 'Old strategy name', 'U0001234') + self.longbridge = instance('longbridge', 'sg', 'sg', 'SG') + self.management = { + 'initialized': True, + 'instances': [instance('longbridge', 'paper', 'paper', 'PAPER'), self.longbridge, self.ibkr], + } + self.roster = [binding('ibkr', 'ibkr-live', 'digest-opaque-target'), + binding('longbridge', 'sg-snapshot', 'sg')] + self.links = [link(self.roster[0], self.ibkr), link(self.roster[1], self.longbridge)] + + def test_uses_config_order_and_verified_account_identity(self): + original = deepcopy(self.roster) + self.ibkr['config']['label'] = 'New strategy name' + result = project_display_roster(self.roster, self.management, self.links) + self.assertEqual(result, [{**original[1], 'account_label': 'SG'}, + {key: value for key, value in original[0].items() if key != 'account_label'}]) + self.assertEqual(self.roster, original) + + def test_updates_configured_display_alias(self): + self.longbridge['config']['label'] = 'Singapore' + self.assertEqual(project_display_roster(self.roster, self.management, self.links)[0]['account_label'], + 'Singapore') + + def test_exports_full_account_only_in_recipient_bound_settings(self): + result = project_telegram_display_settings(self.roster, self.management, self.links, 'recipient') + self.assertNotIn('U0001234', result['DIGEST_EXPECTED_TARGETS_JSON']) + self.assertEqual(json.loads(result['DIGEST_TELEGRAM_IBKR_ACCOUNT_IDS_JSON']), { + 'recipient_id': 'recipient', + 'accounts': [{'candidate_slot': 'ibkr-live', 'opaque_account_uid': 'uid-ibkr-live', + 'target_id': 'digest-opaque-target', 'account_id': 'U0001234'}], + }) + + def test_keeps_platform_together_when_instances_are_interleaved(self): + second = instance('longbridge', 'hk', 'hk', 'HK') + self.management['instances'].append(second) + row = binding('longbridge', 'hk-snapshot', 'hk') + self.roster.append(row) + self.links.append(link(row, second)) + result = project_display_roster(self.roster, self.management, self.links) + self.assertEqual([item.get('account_label') for item in result], ['SG', 'HK', None]) + + def test_changed_management_account_cannot_rename_old_digest_binding(self): + self.ibkr['config']['account_selector'] = 'U0004321' + with self.assertRaisesRegex(ValueError, 'digest_configured_account_identity_changed'): + project_display_roster(self.roster, self.management, self.links) + + def test_wrong_digest_target_and_missing_link_are_rejected(self): + with self.assertRaisesRegex(ValueError, 'digest_enrollment_links_do_not_match_roster'): + project_display_roster(self.roster, self.management, self.links[1:]) + self.roster[0]['target_id'] = 'changed-target' + with self.assertRaisesRegex(ValueError, 'digest_enrollment_links_do_not_match_roster'): + project_display_roster(self.roster, self.management, self.links)