From 0031721e662c5438dc57c74df0aab6dab112c559 Mon Sep 17 00:00:00 2001 From: Pigbibi <20649888+Pigbibi@users.noreply.github.com> Date: Sat, 14 Nov 2026 04:25:35 +0800 Subject: [PATCH 1/7] refactor: split platform entry into route modules Keep main as an explicit named re-export surface for tests and process entry, with routes and wiring moved into dedicated modules. Co-authored-by: Cursor --- http_routes.py | 561 +++++++++++ main.py | 1759 ++++++----------------------------- notifications_runtime.py | 198 ++++ runtime_wiring.py | 268 ++++++ strategy_plugins_runtime.py | 124 +++ 5 files changed, 1437 insertions(+), 1473 deletions(-) create mode 100644 http_routes.py create mode 100644 notifications_runtime.py create mode 100644 runtime_wiring.py create mode 100644 strategy_plugins_runtime.py diff --git a/http_routes.py b/http_routes.py new file mode 100644 index 00000000..b6b3b371 --- /dev/null +++ b/http_routes.py @@ -0,0 +1,561 @@ +"""Flask application, cycle runners, and HTTP route handlers.""" + +from __future__ import annotations + + +import sys + +_FALLBACK_MODULE_NAME = __name__ + + +class _MainModuleProxy: + """Prefer ``main.`` after re-exports; fall back during module import.""" + + def __getattr__(self, name): + main_mod = sys.modules.get("main") + if main_mod is not None and name in getattr(main_mod, "__dict__", {}): + return main_mod.__dict__[name] + return getattr(sys.modules[_FALLBACK_MODULE_NAME], name) + + +main = _MainModuleProxy() + +import dataclasses +import json +import importlib +import os +import time +from datetime import datetime, timezone + +from flask import Flask, request + +import google.auth +import requests +from application.monitor_dispatcher import dispatch_due_monitors, load_monitor_targets +from application.startup_config import ( + StartupConfigReceipt, + normalize_monitor_dispatch_targets, + normalize_plugin_mounts_strategy, + overlay_env, +) +from application.broker_reconciliation import ( + account_snapshot_enabled, + collect_read_only_reconciliation_observations, + reconciliation_enabled, + run_read_only_account_snapshot, + run_read_only_broker_reconciliation, +) +from application.execution_receipt_adapter import ( + attach_cycle_execution_receipt, + attach_terminal_fallback_execution_receipt, +) +from application.runtime_broker_adapters import build_runtime_broker_adapters +from application.runtime_composer import build_runtime_composer +from application.rebalance_service import run_strategy as run_rebalance_cycle +from application.durable_execution_commands import ( + resolve_paper_execution_command_consumer_enabled, +) +from application.paper_execution_command_consumer import ( + consume_due_paper_execution_commands, +) +from application.runtime_strategy_adapters import build_runtime_strategy_adapters +from application.longbridge_execution import submit_order +from application import validation_cycle_observation +from runtime_execution_policy import fractional_buy_execution_enabled, FRACTIONAL_BUY_QUANTITY_STEP +from application.longbridge_portfolio import fetch_strategy_account_state +from application.account_snapshot import ( + begin_natural_cycle_history, cycle_history_observation, end_natural_cycle_history, + read_cycle_account_balance, read_cycle_stock_positions, +) +from scripts.record_daily_account_snapshot import record_projected_daily_account +from entrypoints.cloud_run import is_market_open_now +from runtime_execution_policy import dca_execution_unsupported_reason +from runtime_config_support import load_platform_runtime_settings +from strategy_registry import V7_LONGTERM_COMPOUNDING_PROFILE +from notifications.telegram import build_signal_text, build_strategy_display_name, build_translator +from notifications import unified as unified_notifications +from quant_platform_kit.common.runtime_reports import ( + append_runtime_report_error, + build_runtime_report_base, + finalize_runtime_report, + persist_runtime_report, +) +from quant_platform_kit.common.strategy_plugins import ( + build_strategy_plugin_report_payload, + load_configured_strategy_plugin_signals, + parse_strategy_plugin_mounts, +) +from quant_platform_kit.notifications.strategy_plugin_alerts import ( + StrategyPluginAlertStateSettings, + build_strategy_plugin_alert_context_label as build_alert_context_label, + publish_strategy_plugin_alerts as dispatch_strategy_plugin_alerts, +) +from quant_platform_kit.common.runtime_inputs import build_strategy_evaluation_inputs +from runtime_logging import build_run_id, emit_runtime_log +from quant_platform_kit.longbridge import ( + build_contexts, + calculate_rotation_indicators, + estimate_max_purchase_quantity, + fetch_last_price, + fetch_order_status, + fetch_token_from_secret, + refresh_token_if_needed, +) +from strategy_runtime import load_strategy_runtime +from decision_mapper import map_strategy_decision_to_plan + +app = Flask(__name__) +READ_ONLY_BROKER_RECONCILIATION_COLLECTOR = collect_read_only_reconciliation_observations + +def _summarize_cycle_result_for_report(cycle_result, *, dry_run: bool) -> dict: + if cycle_result is None: + return {'action_done': False, 'order_events_count': 0, 'orders_previewed_count': 0, 'orders_skipped_count': 0, 'notes_count': 0, 'dry_run_order_preview_available': False} + logs = tuple(getattr(cycle_result, 'logs', ()) or ()) + skip_logs = tuple(getattr(cycle_result, 'skip_logs', ()) or ()) + note_logs = tuple(getattr(cycle_result, 'note_logs', ()) or ()) + dry_run_orders = tuple(getattr(cycle_result, 'dry_run_orders', ()) or ()) + pending_orders = tuple(getattr(cycle_result, 'pending_orders', ()) or ()) + quote_snapshots = tuple(getattr(cycle_result, 'quote_snapshots', ()) or ()) + order_events_count = 0 if pending_orders else len(logs) + orders_previewed_count = len(dry_run_orders) if dry_run_orders else order_events_count if dry_run else 0 + broker_submission_done = bool(getattr(cycle_result, 'action_done', False)) + summary = {'action_done': broker_submission_done and (not pending_orders), 'broker_submission_done': broker_submission_done, 'execution_status': 'pending_reconciliation' if pending_orders else 'previewed' if dry_run and broker_submission_done else 'no_action', 'order_events_count': order_events_count, 'orders_pending_count': len(pending_orders), 'orders_previewed_count': orders_previewed_count, 'orders_skipped_count': len(skip_logs), 'notes_count': len(note_logs), 'dry_run_order_preview_available': bool(dry_run and orders_previewed_count > 0)} + if dry_run_orders: + summary['orders_previewed'] = [dict(order) for order in dry_run_orders] + if pending_orders: + summary['orders_pending'] = [dict(order) for order in pending_orders] + if quote_snapshots: + summary['quote_snapshot'] = {'quotes': [dict(snapshot) for snapshot in quote_snapshots]} + execution = dict(getattr(cycle_result, 'execution', {}) or {}) + account_snapshot = execution.get('heartbeat_account_snapshot') + if isinstance(account_snapshot, dict): + summary['heartbeat_account_snapshot'] = dict(account_snapshot) + if execution.get('execution_status') == 'blocked': + summary['execution_status'] = 'blocked' + summary['blocked_reason'] = 'durable_live_execution_command_binding_invalid' if execution.get('blocked_reason') == 'durable_live_execution_command_binding_invalid' else 'unknown' + for field in ('signal_date', 'effective_date', 'execution_timing_contract', 'execution_calendar_source'): + if execution.get(field) not in (None, ''): + summary[field] = execution[field] + durable_command = execution.get('durable_execution_command') + if isinstance(durable_command, dict): + summary['durable_execution_command'] = dict(durable_command) + live_command = execution.get('durable_live_execution_command') + if isinstance(live_command, dict): + summary['durable_live_execution_command'] = dict(live_command) + account_identity = execution.get('account_identity') + if isinstance(account_identity, dict): + summary['account_identity'] = dict(account_identity) + _, summary['notification_digest'] = main.unified_notifications.cycle_facts(cycle_result, dry_run=dry_run) + return summary + +def _min_order_notional_usd() -> float: + return float(getattr(main.RUNTIME_SETTINGS, 'min_order_notional_usd', main.DEFAULT_MIN_ORDER_NOTIONAL_USD)) + +def run_strategy(*, force_run: bool=False, validation_only: bool=False, validation_label: str='backfill'): + if main._v7_application_candidate_bound() and (not validation_only): + print(f'[{main.datetime.now()}] V7 paper candidate is paused; skip all execution paths.', flush=True) + return True + if not validation_only and (not force_run) and (not getattr(main.RUNTIME_SETTINGS, 'runtime_target_enabled', True)): + print(f'[{main.datetime.now()}] Runtime target disabled; skip strategy execution.', flush=True) + return True + composer = main.build_composer(dry_run_only_override=True if validation_only else None) + observation = main.validation_cycle_observation.current() if validation_only else None + if observation is not None: + composer = main.validation_cycle_observation.instrument_composer(composer, observation) + reporting_adapters = composer.build_reporting_adapters() + log_context, report = reporting_adapters.start_run() + notification_delivery_events: list[dict] = [] + try: + notification_adapters = composer.build_notification_adapters(delivery_events=notification_delivery_events) + except TypeError as exc: + if 'delivery_events' not in str(exc): + raise + notification_adapters = composer.build_notification_adapters() + strategy_plugin_signals, strategy_plugin_error = composer.load_strategy_plugin_signals(getattr(main.RUNTIME_SETTINGS, 'strategy_plugin_mounts_json', None)) + composer.attach_strategy_plugin_report(report, signals=strategy_plugin_signals, error=strategy_plugin_error) + failure_phase = 'market_hours' + try: + reporting_adapters.log_event(log_context, 'strategy_cycle_started', message='Starting strategy execution') + print(composer.with_prefix(f'[{main.datetime.now()}] Starting strategy...'), flush=True) + market_open = main.is_market_open_now(calendar_name=main.MARKET_CALENDAR, timezone_name=main.MARKET_TIMEZONE) + if isinstance(market_open, tuple): + market_open, error = market_open + if error is not None: + reporting_adapters.log_event(log_context, 'market_hours_check_failed', message='Market hours check failed', severity='WARNING', error_message=str(error), market=main.MARKET, market_calendar=main.MARKET_CALENDAR, market_timezone=main.MARKET_TIMEZONE) + print(composer.with_prefix(f'Market hours check failed: {error}'), flush=True) + if not market_open and (not force_run): + reporting_adapters.log_event(log_context, 'outside_market_hours', message='Outside market hours; skip execution', market=main.MARKET, market_calendar=main.MARKET_CALENDAR, market_timezone=main.MARKET_TIMEZONE) + main.finalize_runtime_report(report, status='skipped', diagnostics={'skip_reason': 'market_closed'}) + print(composer.with_prefix('Outside market hours; skip.'), flush=True) + return True + if force_run and (not market_open): + reporting_adapters.log_event(log_context, 'market_hours_bypassed', message=f'Market hours bypassed for {validation_label} execution', market=main.MARKET, market_calendar=main.MARKET_CALENDAR, market_timezone=main.MARKET_TIMEZONE) + print(composer.with_prefix(f'Market hours bypassed for {validation_label} verification; validation only, no orders will be submitted.'), flush=True) + unsupported_reason = main.dca_execution_unsupported_reason(main.STRATEGY_PROFILE) + if unsupported_reason is not None: + reporting_adapters.log_event(log_context, 'strategy_execution_unsupported', message='Strategy requires fractional-share execution; skip', skip_reason=unsupported_reason, strategy_profile=main.STRATEGY_PROFILE, market=main.MARKET, market_calendar=main.MARKET_CALENDAR, market_timezone=main.MARKET_TIMEZONE) + main.finalize_runtime_report(report, status='skipped', diagnostics={'skip_reason': unsupported_reason}) + print(composer.with_prefix(f'Strategy {main.STRATEGY_PROFILE} requires fractional-share execution; skip.'), flush=True) + return True + if not validation_only: + main.publish_strategy_plugin_alerts(strategy_plugin_signals, report=report) + failure_phase = 'runtime_configuration' + try: + rebalance_runtime = composer.build_rebalance_runtime(silent_cycle_notifications=validation_only, notification_delivery_events=notification_delivery_events) + except TypeError as exc: + if 'notification_delivery_events' not in str(exc): + raise + rebalance_runtime = composer.build_rebalance_runtime(silent_cycle_notifications=validation_only) + if observation is not None: + validation_bootstrap = getattr(rebalance_runtime.bootstrap, 'validation_cycle', None) + if not callable(validation_bootstrap): + raise RuntimeError('validation_read_only_bootstrap_unavailable') + rebalance_runtime = main.dataclasses.replace(rebalance_runtime, bootstrap=validation_bootstrap, notify_issue=lambda _title, _detail: None) + rebalance_config = composer.build_rebalance_config(strategy_plugin_signals=strategy_plugin_signals, strategy_plugin_error=strategy_plugin_error, notification_title_key='dry_run_title' if validation_only and validation_label == 'dry_run' else '', cash_only_execution=main.CASH_ONLY_EXECUTION, live_execution_session_authorized=bool(market_open and (not validation_only))) + if observation is not None: + main.validation_cycle_observation.require_matching_journal_source(rebalance_config, reporting_adapters) + observation.report_base_dir = getattr(reporting_adapters, 'report_base_dir', None) + failure_phase = 'strategy_cycle' + if observation is not None: + observation.cycle_entered = True + with main.unified_notifications.bind_report(report, validation_only=validation_only): + cycle_result = main.run_rebalance_cycle(runtime=rebalance_runtime, config=rebalance_config) + if observation is not None: + observation.cycle_returned = True + failure_phase = 'reporting' + signal_snapshot = {} + if cycle_result is not None: + execution = dict(getattr(cycle_result, 'execution', {}) or {}) + signal_snapshot = dict(execution.get('signal_snapshot') or {}) + execution_summary = main._summarize_cycle_result_for_report(cycle_result, dry_run=bool(report.get('dry_run'))) + if 'routine_observation_counts' in (report.get('summary') or {}): + execution_summary['routine_observation_counts'] = report['summary']['routine_observation_counts'] + if not validation_only: + try: + main.unified_notifications.attach_cycle_event(execution_summary, cycle_result, report=report, config=rebalance_config) + except Exception: + print('notification_event_attachment_unconfirmed', flush=True) + main.unified_notifications.record_failure('notification_event_attachment_unconfirmed', report=report) + notification_delivery_log = main._build_notification_delivery_log_for_report(platform='longbridge', strategy_profile=main.STRATEGY_PROFILE, run_id=str(report.get('run_id') or ''), dry_run=bool(report.get('dry_run')), orders_previewed_count=int(execution_summary.get('orders_previewed_count') or 0), delivery_events=notification_delivery_events) + if notification_delivery_log: + execution_summary['notification_delivery_log'] = notification_delivery_log + if main.unified_notifications.enabled(): + notification_delivery_events.extend(({'sink': 'notification_store', 'delivery_status': 'failed', 'transport_acknowledged': False, 'error_type': code} for code in report.get('notification_unified_errors', ()))) + notification_delivery_summary = main._build_notification_delivery_summary(notification_delivery_events) + if notification_delivery_summary: + execution_summary['notification_delivery_summary'] = notification_delivery_summary + if signal_snapshot: + reporting_adapters.log_event(log_context, 'strategy_signal_snapshot', message='Strategy signal snapshot', **signal_snapshot) + try: + main.attach_cycle_execution_receipt(report, cycle_result) + except ValueError: + pass + main.finalize_runtime_report(report, status='ok', summary=execution_summary, diagnostics={'signal_snapshot': signal_snapshot} if signal_snapshot else None) + reporting_adapters.log_event(log_context, 'strategy_cycle_completed', message='Strategy execution completed') + return True + except Exception as exc: + failure_reason = main._runtime_failure_reason(exc) + main.append_runtime_report_error(report, stage='strategy_cycle', message='strategy_cycle_failed', error_type=type(exc).__name__) + reporting_adapters.log_event(log_context, 'strategy_cycle_failed', message='Strategy execution failed', severity='ERROR', error_type=type(exc).__name__, error_message='strategy_cycle_failed', failure_phase=failure_phase, failure_reason=failure_reason) + if observation is None and (not (validation_only and main._v7_application_candidate_bound())): + try: + + def publish_failure(): + return notification_adapters.publish_cycle_notification(detailed_text=main._compact_error_notification(exc, phase=failure_phase), compact_text=main._compact_error_notification(exc, phase=failure_phase)) + main.unified_notifications.dispatch_immediate(publish_failure, kind='real_error', incident=f'runtime_failure:{failure_phase}:{failure_reason}', report=report, actually_ran=True) + except Exception as notification_exc: + notification_delivery_events.append({'sink': 'telegram', 'delivery_status': 'failed', 'transport_acknowledged': False, 'error_type': type(notification_exc).__name__}) + reporting_adapters.log_event(log_context, 'strategy_error_notification_failed', message='Strategy error notification failed', severity='ERROR', error_type=type(notification_exc).__name__) + error_summary = {} + notification_delivery_summary = main._build_notification_delivery_summary(notification_delivery_events) + if notification_delivery_summary: + error_summary['notification_delivery_summary'] = notification_delivery_summary + main.finalize_runtime_report(report, status='error', summary=error_summary or None, diagnostics={'failure_phase': failure_phase, 'failure_reason': failure_reason}) + return False + finally: + if 'execution_receipt' not in report: + try: + main.attach_terminal_fallback_execution_receipt(report) + except ValueError: + pass + if observation is not None: + status = str(report.get('status') or '').strip().lower() + observation.report_status = status if status in {'ok', 'error', 'skipped'} else 'unknown' + observation.execution_receipt_present = isinstance(report.get('execution_receipt'), dict) + if observation is None and (not (validation_only and main._v7_application_candidate_bound())): + try: + report_path = reporting_adapters.persist_execution_report(report) + print(f'execution_report {report_path}', flush=True) + if not validation_only and report.get('status') in {'ok', 'completed', 'success'}: + if not main.unified_notifications.report_archive_confirmed(report, report_path, reporting_adapters.report_gcs_prefix_uri): + raise ValueError('notification_report_archive_unconfirmed') + if main.unified_notifications.persist_archived_report(report, env=main.os.environ) is not True: + raise ValueError('notification_archive_candidate_unconfirmed') + if observation is None and main.unified_notifications.enabled(): + main.unified_notifications.recover_runtime(report=report) + except Exception as persist_exc: + print(f'failed to persist execution report: {type(persist_exc).__name__}', flush=True) + if not validation_only and report.get('status') in {'ok', 'completed', 'success'}: + main.unified_notifications._storage_failure(lambda title, detail: rebalance_runtime.notifications.send_text(f'{title}\n{detail}')) + +def _bounded_probe_error_code(exc: Exception) -> int | None: + try: + code = getattr(exc, 'code', None) + except Exception: + return None + if type(code) is not int or not -2 ** 31 <= code <= 2 ** 31 - 1: + return None + return code + +def run_probe(*, response_body: str='Probe OK'): + if main._v7_application_candidate_bound(): + return (response_body, 200) + history_scope = str(main.ACCOUNT_REGION or '').strip() + record_account_history = history_scope in {'HK', 'SG'} and str(main.os.getenv('ACCOUNT_HISTORY_RECORDING_ENABLED') or '').strip() == 'true' + history_capture = main.begin_natural_cycle_history() if record_account_history else None + composer = None + reporting_adapters = None + log_context = None + report = None + probe_step = 'settings' + try: + composer = main.build_composer(dry_run_only_override=True) + reporting_adapters = composer.build_reporting_adapters() + log_context, report = reporting_adapters.start_run() + reporting_adapters.log_event(log_context, 'health_probe_received', message='Received health probe request', execution_window='probe') + probe_step = 'context' + runtime = composer.build_rebalance_runtime(silent_cycle_notifications=True) + if record_account_history: + probe_step = 'context' + quote_context, trade_context, source_binding = composer.build_account_snapshot_broker_contexts() + probe_step = 'balance' + main.read_cycle_account_balance(lambda: trade_context.account_balance(currency='USD')) + stock_reader = getattr(trade_context, 'stock_positions', None) + if callable(stock_reader): + try: + main.read_cycle_stock_positions(stock_reader) + except Exception: + pass + observation = main.cycle_history_observation() + if observation is None: + raise RuntimeError('account history observation incomplete') + probe_step = 'archive' + result = main.record_projected_daily_account(main.os.environ, account_scope=history_scope, source_binding=source_binding, balances=observation['projection']['broker_reported_balances'], cash=observation['projection']['cash'], started=observation['started'], finished=observation['finished'], open_store=main._open_account_history_store, financing=observation['projection'].get('financing'), positions=observation['projection'].get('broker_reported_positions')) + print(f"account_history status={result.status} category={result.category or '-'}", flush=True) + if result.status != 'recorded': + raise RuntimeError('account history write was not confirmed') + probe_step = 'indicators' + _indicators = composer.strategy_adapters.calculate_strategy_indicators(quote_context) + else: + probe_step = 'bootstrap' + quote_context, trade_context, _indicators = runtime.bootstrap() + probe_step = 'portfolio_snapshot' + snapshot = runtime.portfolio_port_factory(quote_context, trade_context).get_portfolio_snapshot() + positions = tuple(getattr(snapshot, 'positions', ()) or ()) + buying_power = float(getattr(snapshot, 'buying_power', 0.0) or 0.0) + total_equity = float(getattr(snapshot, 'total_equity', 0.0) or 0.0) + main.finalize_runtime_report(report, status='ok', summary={'buying_power': buying_power, 'total_equity': total_equity, 'positions_count': len(positions)}) + reporting_adapters.log_event(log_context, 'health_probe_completed', message='Health probe completed', execution_window='probe', buying_power=buying_power, total_equity=total_equity, positions_count=len(positions)) + return (response_body, 200) + except Exception as exc: + error_code = main._bounded_probe_error_code(exc) + error_diagnostics = {'probe_step': probe_step, 'error_code_status': 'known' if error_code is not None else 'unknown'} + if error_code is not None: + error_diagnostics['error_code'] = error_code + if report is not None: + main.append_runtime_report_error(report, stage='health_probe', message='health_probe_failed', error_type=type(exc).__name__, **error_diagnostics) + main.finalize_runtime_report(report, status='error') + if reporting_adapters is not None and log_context is not None: + reporting_adapters.log_event(log_context, 'health_probe_failed', message='Health probe failed', severity='ERROR', execution_window='probe', error_type=type(exc).__name__, error_message='health_probe_failed', **error_diagnostics) + err = main._compact_error_notification(exc, title=main.t('health_probe_title'), prefix=main.t('health_probe_error_prefix'), phase='health_probe') + if composer is not None: + + def publish_probe_failure(): + return composer.build_notification_adapters().publish_cycle_notification(detailed_text=err, compact_text=main._compact_error_notification(exc, title=main.t('health_probe_title'), prefix=main.t('health_probe_error_prefix'), phase='health_probe')) + main.unified_notifications.dispatch_immediate(publish_probe_failure, kind='real_error', incident='runtime_failure:health_probe', report=report) + else: + print(err, flush=True) + return ('Error', 500) + finally: + if history_capture is not None: + main.end_natural_cycle_history(history_capture) + try: + if reporting_adapters is not None and report is not None: + report_path = reporting_adapters.persist_execution_report(report) + print(f'execution_report {report_path}', flush=True) + if main.unified_notifications.enabled(): + main.unified_notifications.recover_runtime(report=report, probe=True) + except Exception as persist_exc: + print(f'failed to persist execution report: {type(persist_exc).__name__}', flush=True) + +def _paper_command_consumer_session_date() -> str: + """Return the exchange-local session date used to claim due commands.""" + try: + import pytz + return main.datetime.now(pytz.timezone(main.MARKET_TIMEZONE)).date().isoformat() + except Exception: + return main.datetime.now().date().isoformat() + +def _paper_command_consumer_runtime_is_isolated() -> bool: + runtime_target = getattr(main.RUNTIME_SETTINGS, 'runtime_target', None) + return bool(getattr(main.RUNTIME_SETTINGS, 'dry_run_only', False) and runtime_target is not None and (str(getattr(runtime_target, 'execution_mode', '') or '').lower() == 'paper')) + +def run_paper_execution_command_consumer() -> bool: + """Run an explicitly invoked, read-only paper command consumer. + + This handler never constructs an execution port or calls the LongBridge + order API. It is deliberately separate from ``/run`` so a normal cycle + cannot start consuming delayed commands by accident. + """ + if main._v7_application_candidate_bound(): + return True + if not main._paper_command_consumer_runtime_is_isolated(): + raise RuntimeError('paper command consumer requires RUNTIME_TARGET_JSON.execution_mode=paper and LONGBRIDGE_DRY_RUN_ONLY=true') + if not main.resolve_paper_execution_command_consumer_enabled(env_reader=main.os.getenv, dry_run_only=bool(main.RUNTIME_SETTINGS.dry_run_only)): + raise RuntimeError('paper command consumer is not enabled') + composer = main.build_composer() + config = composer.build_rebalance_config() + reporting_adapters = composer.build_reporting_adapters() + log_context, report = reporting_adapters.start_run() + try: + reporting_adapters.log_event(log_context, 'paper_execution_command_consumer_started', message='Starting read-only paper execution command consumer') + quote_context, trade_context = composer.build_read_only_broker_contexts() + portfolio = composer.broker_adapters.build_portfolio_port(quote_context, trade_context).get_portfolio_snapshot() + result = main.consume_due_paper_execution_commands(store=config.execution_command_store, as_of_session=main._paper_command_consumer_session_date(), claimant=str(main.os.getenv('K_SERVICE') or 'longbridge-paper-command-consumer'), portfolio=portfolio, market_data_port=composer.broker_adapters.build_market_data_port(quote_context), runtime_release_receipt=config.runtime_release_receipt, expected_strategy_release=config.expected_strategy_release, expected_command_binding={'platform': 'longbridge', 'account_scope': str(config.execution_state_account_scope or 'unknown'), 'strategy_profile': str(config.strategy_profile or 'unknown')}) + report_status = 'ok' if result.get('status') == 'ok' else 'skipped' + main.finalize_runtime_report(report, status=report_status, summary={'paper_execution_command_consumer': result}) + reporting_adapters.log_event(log_context, 'paper_execution_command_consumer_completed', message='Paper execution command consumer completed', result_status=result.get('status'), commands_count=len(tuple(result.get('commands') or ()))) + return True + except Exception as exc: + main.append_runtime_report_error(report, stage='paper_execution_command_consumer', message=str(exc), error_type=type(exc).__name__) + main.finalize_runtime_report(report, status='error') + reporting_adapters.log_event(log_context, 'paper_execution_command_consumer_failed', message='Paper execution command consumer failed', severity='ERROR', error_type=type(exc).__name__, error_message=str(exc)) + return False + finally: + try: + report_path = reporting_adapters.persist_execution_report(report) + print(f'execution_report {report_path}', flush=True) + except Exception as persist_exc: + print(f'failed to persist execution report: {persist_exc}', flush=True) + +def run_broker_reconciliation(): + """Run the explicitly enabled, order-port-free reconciliation boundary.""" + return main.run_read_only_broker_reconciliation(enabled=main.reconciliation_enabled(main.os.getenv), account_scope=getattr(main.RUNTIME_SETTINGS, 'account_region', None), runtime_target=getattr(main.RUNTIME_SETTINGS, 'runtime_target', None), strategy_profile=main.STRATEGY_PROFILE, project_id=main.PROJECT_ID, build_read_only_contexts=lambda: main.build_composer().build_read_only_broker_contexts(), collect_evidence=main.READ_ONLY_BROKER_RECONCILIATION_COLLECTOR) + +def _open_account_history_store(project_id): + from quant_platform_kit.cloud import get_object_store + return get_object_store(project_id=project_id) + +def run_account_snapshot(): + """Read bounded account facts through the order-port-free broker contexts.""" + return main.run_read_only_account_snapshot(enabled=main.account_snapshot_enabled(main.os.getenv), account_scope=getattr(main.RUNTIME_SETTINGS, 'account_region', None), build_read_only_contexts=lambda: main.build_composer().build_account_snapshot_broker_contexts(), collect_evidence=main.READ_ONLY_BROKER_RECONCILIATION_COLLECTOR) + +@app.route('/run', methods=['POST']) +def handle_trigger(): + """Entrypoint for Cloud Run / scheduler: run strategy and return 200.""" + if main.request_method() != 'POST': + return ('Method Not Allowed', 405) + return main._route_with_runtime_error_fallback(main.run_strategy, success_body='OK', route_label='POST /run') + +@app.route('/backfill', methods=['POST', 'GET']) +def handle_backfill(): + """Manual backfill entrypoint for verification-only execution.""" + + def run_validation(): + return main._route_with_runtime_error_fallback(main.run_strategy, force_run=True, validation_only=True, success_body='OK', route_label='POST /backfill') + if str(main.ACCOUNT_REGION or '').strip().upper() != 'PAPER': + return run_validation() + headers = getattr(main.request, 'headers', {}) + with main.validation_cycle_observation.observe(headers.get('X-Cloud-Trace-Context'), main.PROJECT_ID): + return run_validation() + +@app.route('/dry-run', methods=['POST', 'GET']) +def handle_dry_run(): + """Strategy dry-run entrypoint.""" + + def run_validation(): + return main._route_with_runtime_error_fallback(main.run_strategy, force_run=True, validation_only=True, validation_label='dry_run', success_body='Dry Run OK', route_label='POST /dry-run') + if str(main.ACCOUNT_REGION or '').strip().upper() != 'PAPER': + return run_validation() + headers = getattr(main.request, 'headers', {}) + with main.validation_cycle_observation.observe(headers.get('X-Cloud-Trace-Context'), main.PROJECT_ID): + return run_validation() + +@app.route('/probe', methods=['POST']) +def handle_probe(): + """Post-open broker/account health probe; notify only on failure.""" + if main.request_method() != 'POST': + return ('Method Not Allowed', 405) + return main._route_with_runtime_error_fallback(main.run_probe, success_body='Probe OK', route_label='POST /probe') + +@app.route('/paper-command-consumer', methods=['POST']) +def handle_paper_execution_command_consumer(): + """Explicit paper-only command verification endpoint; never scheduler-routed.""" + return main._route_with_runtime_error_fallback(main.run_paper_execution_command_consumer, success_body='Paper command consumer OK', route_label='POST /paper-command-consumer') + +@app.route('/reconcile', methods=['POST']) +def handle_broker_reconciliation(): + """Return only redacted QPK E3 evidence; never invoke normal execution.""" + if main.request_method() != 'POST': + return ('Method Not Allowed', 405) + payload, status = main.run_broker_reconciliation() + return (main.json.dumps(payload, ensure_ascii=True), status, {'Content-Type': 'application/json'}) + +@app.route('/account-snapshot', methods=['GET']) +def handle_account_snapshot(): + """Return a non-cacheable, read-only and authority-free account snapshot.""" + if main.request_method() != 'GET': + return ('Method Not Allowed', 405) + payload, status = main.run_account_snapshot() + return (main.json.dumps(payload, ensure_ascii=True), status, {'Content-Type': 'application/json', 'Cache-Control': 'no-store'}) + +@app.route('/v7-paper-application-loaded', methods=['GET']) +def handle_v7_paper_application_loaded(): + """Read back the disabled V7 process identity without broker access.""" + if main.request_method() != 'GET': + return ('Method Not Allowed', 405) + from application.v7_paper_application import V7PaperApplicationError, build_v7_loaded_version_readback, load_v7_paper_application_binding + try: + binding = load_v7_paper_application_binding(main.os.environ) + if binding is None: + return (main.json.dumps({'status': 'unavailable', 'reason': 'application_binding_missing'}, ensure_ascii=True), 409, {'Content-Type': 'application/json', 'Cache-Control': 'no-store'}) + payload = build_v7_loaded_version_readback(binding, runtime_settings=main.RUNTIME_SETTINGS, strategy_runtime=main.STRATEGY_RUNTIME, runtime_env=main.os.environ) + except V7PaperApplicationError: + return (main.json.dumps({'status': 'uncertain', 'reason': 'loaded_version_unavailable'}, ensure_ascii=True), 409, {'Content-Type': 'application/json', 'Cache-Control': 'no-store'}) + return (main.json.dumps({'status': 'applied_paused', 'readback': payload}, ensure_ascii=True), 200, {'Content-Type': 'application/json', 'Cache-Control': 'no-store'}) + +def _v7_application_candidate_bound() -> bool: + """Keep the exact V7 binding paused even when a route bypasses the flag.""" + if main.STRATEGY_PROFILE != main.V7_LONGTERM_COMPOUNDING_PROFILE: + return False + from application.v7_paper_application import load_v7_paper_application_binding + return load_v7_paper_application_binding(main.os.environ) is not None + +@app.route('/monitor-dispatch', methods=['POST', 'GET']) +def handle_monitor_dispatch(): + if main.request_method() == 'GET': + return ('Monitor Dispatch OK - use POST to dispatch due monitor checks', 200) + try: + result = main.dispatch_due_monitors(main._load_monitor_dispatch_targets()) + except Exception as exc: + return main._handle_route_runtime_error(exc, route_label='POST /monitor-dispatch') + return (main.json.dumps(result, ensure_ascii=False), 200, {'Content-Type': 'application/json'}) + +def request_method() -> str: + try: + return main.request.method + except Exception: + return 'GET' + +@app.route('/healthz', methods=['GET']) +@app.route('/health', methods=['GET']) +def health(): + critical_errors: list[str] = [] + for module_path in ('application.runtime_composer', 'application.runtime_reporting_adapters', 'application.runtime_strategy_adapters', 'runtime_config_support', 'runtime_execution_policy'): + try: + main.importlib.import_module(module_path) + except Exception as exc: + critical_errors.append(f'{module_path}: {type(exc).__name__}: {exc}') + if critical_errors: + return (main.json.dumps({'status': 'unhealthy', 'errors': critical_errors}, ensure_ascii=False), 500) + return ('OK', 200) diff --git a/main.py b/main.py index cf8eacb5..02093335 100644 --- a/main.py +++ b/main.py @@ -1,8 +1,18 @@ -""" -LongPort Cloud Run strategy executor. -Runs the configured shared UsEquityStrategies profile for the deployed LongBridge account. -Runs on Cloud Run; token from Secret Manager, orders via LongPort OpenAPI, alerts via Telegram. -""" +"""Platform HTTP entrypoint with explicit named re-exports.""" + +from __future__ import annotations + +import sys + +for _split_module_name in ( + "runtime_wiring", + "notifications_runtime", + "strategy_plugins_runtime", + "http_routes", +): + sys.modules.pop(_split_module_name, None) +del _split_module_name + import dataclasses import json import importlib @@ -87,1483 +97,286 @@ from strategy_runtime import load_strategy_runtime from decision_mapper import map_strategy_decision_to_plan -app = Flask(__name__) - -# This collector is limited to documented LongPort read endpoints and is -# independently fail-closed for incomplete account identity/order coverage. -READ_ONLY_BROKER_RECONCILIATION_COLLECTOR = collect_read_only_reconciliation_observations - -# --------------------------------------------------------------------------- -# Config and constants (GCP project, Telegram, execution and strategy params) -# --------------------------------------------------------------------------- -def get_project_id(): - try: - _, project_id = google.auth.default() - return project_id - except Exception: - return os.getenv("GOOGLE_CLOUD_PROJECT") - -RUNTIME_SETTINGS = load_platform_runtime_settings(project_id_resolver=get_project_id) -PROJECT_ID = RUNTIME_SETTINGS.project_id -SECRET_NAME = RUNTIME_SETTINGS.secret_name -ACCOUNT_PREFIX = RUNTIME_SETTINGS.account_prefix -STRATEGY_PROFILE = RUNTIME_SETTINGS.strategy_profile -STRATEGY_DISPLAY_NAME = RUNTIME_SETTINGS.strategy_display_name - - -def _normalize_plugin_mounts_strategy(raw_mounts_json: str | None) -> str | None: - normalized, _corrections = normalize_plugin_mounts_strategy( - raw_mounts_json, - strategy_profile=STRATEGY_PROFILE, - ) - return normalized - - -def _adopt_startup_config(settings, env): - """Explicitly normalize startup config without mutating ``os.environ`` (B08). - - Returns ``(settings, monitor_dispatch_env_overrides, receipt)``. - """ - corrections = [] - if hasattr(settings, "strategy_plugin_mounts_json"): - normalized, plugin_corrections = normalize_plugin_mounts_strategy( - settings.strategy_plugin_mounts_json, - strategy_profile=STRATEGY_PROFILE, - ) - corrections.extend(plugin_corrections) - if normalized != settings.strategy_plugin_mounts_json: - if dataclasses.is_dataclass(settings) and not isinstance(settings, type): - settings = dataclasses.replace(settings, strategy_plugin_mounts_json=normalized) - else: # non-dataclass settings objects (test doubles) - setattr(settings, "strategy_plugin_mounts_json", normalized) - overrides, monitor_corrections = normalize_monitor_dispatch_targets( - env, - strategy_profile=STRATEGY_PROFILE, - ) - corrections.extend(monitor_corrections) - receipt = StartupConfigReceipt( - strategy_profile=STRATEGY_PROFILE, - corrections=tuple(corrections), - env_overrides=tuple(sorted(overrides)), - ) - return settings, overrides, receipt - - -RUNTIME_SETTINGS, MONITOR_DISPATCH_ENV_OVERRIDES, STARTUP_CONFIG_RECEIPT = _adopt_startup_config( +from runtime_wiring import ( + get_project_id, RUNTIME_SETTINGS, - os.environ, -) - - -def _load_monitor_dispatch_targets(): - # No corrections: identical call to before. Corrections: read through an - # explicit overlay instead of the previously mutated ``os.environ``. - if not MONITOR_DISPATCH_ENV_OVERRIDES: - return load_monitor_targets() - return load_monitor_targets(overlay_env(os.environ, MONITOR_DISPATCH_ENV_OVERRIDES)) - - -ACCOUNT_REGION = RUNTIME_SETTINGS.account_region -MARKET = RUNTIME_SETTINGS.market -MARKET_CALENDAR = RUNTIME_SETTINGS.market_calendar -MARKET_TIMEZONE = RUNTIME_SETTINGS.market_timezone -SYMBOL_SUFFIX = RUNTIME_SETTINGS.symbol_suffix -TRADING_CURRENCY = RUNTIME_SETTINGS.trading_currency -NOTIFY_LANG = RUNTIME_SETTINGS.notify_lang -TG_TOKEN = RUNTIME_SETTINGS.tg_token -TG_CHAT_ID = RUNTIME_SETTINGS.tg_chat_id -from quant_platform_kit.notifications.cycle_channel import resolve_cycle_channel_and_url - -_NOTIFICATION_CHANNEL, _NOTIFICATION_WEBHOOK_URL = resolve_cycle_channel_and_url( - explicit_channel=RUNTIME_SETTINGS.notification_channel, - telegram_token=TG_TOKEN, - telegram_chat_id=TG_CHAT_ID, - wecom_url=RUNTIME_SETTINGS.wecom_webhook_url, - dingtalk_url=RUNTIME_SETTINGS.dingtalk_webhook_url, - feishu_url=RUNTIME_SETTINGS.feishu_webhook_url, - serverchan_url=RUNTIME_SETTINGS.serverchan_webhook_url, -) -CASH_ONLY_EXECUTION = getattr(RUNTIME_SETTINGS, "cash_only_execution", True) -STRATEGY_RUNTIME = load_strategy_runtime( + PROJECT_ID, + SECRET_NAME, + ACCOUNT_PREFIX, STRATEGY_PROFILE, - runtime_settings=RUNTIME_SETTINGS, - logger=lambda message: print(message, flush=True), -) -STRATEGY_RUNTIME_CONFIG = dict(STRATEGY_RUNTIME.merged_runtime_config) -MANAGED_SYMBOLS = STRATEGY_RUNTIME.managed_symbols -AVAILABLE_INPUTS = frozenset(STRATEGY_RUNTIME.runtime_adapter.available_inputs) -BENCHMARK_SYMBOL = str(STRATEGY_RUNTIME_CONFIG.get("benchmark_symbol", "QQQ")) - - -def _emit_v7_loaded_version_log() -> None: - """Emit one safe startup receipt for the disabled candidate revision.""" - - if STRATEGY_PROFILE != V7_LONGTERM_COMPOUNDING_PROFILE: - return - from application.v7_paper_application import ( - V7PaperApplicationError, - build_v7_loaded_version_readback, - load_v7_paper_application_binding, - ) - - binding = load_v7_paper_application_binding(os.environ) - if binding is None: - return - try: - readback = build_v7_loaded_version_readback( - binding, - runtime_settings=RUNTIME_SETTINGS, - strategy_runtime=STRATEGY_RUNTIME, - runtime_env=os.environ, - ) - print( - json.dumps( - {"event": "v7_paper_application_loaded", "status": "applied_paused", "readback": readback}, - ensure_ascii=True, - separators=(",", ":"), - ), - flush=True, - ) - except V7PaperApplicationError: - print( - json.dumps( - {"event": "v7_paper_application_loaded", "status": "uncertain"}, - ensure_ascii=True, - separators=(",", ":"), - ), - flush=True, - ) - - -_emit_v7_loaded_version_log() -SIGNAL_EFFECTIVE_AFTER_TRADING_DAYS = getattr( - getattr(STRATEGY_RUNTIME.runtime_adapter, "runtime_policy", None), - "signal_effective_after_trading_days", - None, + STRATEGY_DISPLAY_NAME, + _normalize_plugin_mounts_strategy, + _adopt_startup_config, + MONITOR_DISPATCH_ENV_OVERRIDES, + STARTUP_CONFIG_RECEIPT, + _load_monitor_dispatch_targets, + ACCOUNT_REGION, + MARKET, + MARKET_CALENDAR, + MARKET_TIMEZONE, + SYMBOL_SUFFIX, + TRADING_CURRENCY, + NOTIFY_LANG, + TG_TOKEN, + TG_CHAT_ID, + _NOTIFICATION_CHANNEL, + _NOTIFICATION_WEBHOOK_URL, + CASH_ONLY_EXECUTION, + STRATEGY_RUNTIME, + STRATEGY_RUNTIME_CONFIG, + MANAGED_SYMBOLS, + AVAILABLE_INPUTS, + BENCHMARK_SYMBOL, + _emit_v7_loaded_version_log, + SIGNAL_EFFECTIVE_AFTER_TRADING_DAYS, + LIMIT_SELL_DISCOUNT, + LIMIT_BUY_PREMIUM, + DEFAULT_LIMIT_BUY_PREMIUM_BY_SYMBOL, + ORDER_POLL_INTERVAL_SEC, + ORDER_POLL_MAX_ATTEMPTS, + TOKEN_REFRESH_THRESHOLD_DAYS, + DEFAULT_SAFE_HAVEN_CASH_SUBSTITUTE_THRESHOLD_USD, + DEFAULT_MIN_ORDER_NOTIONAL_USD, + _FRACTIONAL_BUY_EXECUTION, + _profile_submit_order, + _profile_estimate_max_purchase_quantity, + SEPARATOR, + COMPACT_ERROR_NOTIFICATION_MAX_CHARS, + _load_limit_buy_premium_by_symbol, + LIMIT_BUY_PREMIUM_BY_SYMBOL, + t, + _build_notification_delivery_log_for_report, + _build_notification_delivery_summary, + signal_text, + strategy_display_name, + log_position_snapshot, + log_runtime_warning, + BROKER_ADAPTERS, + STRATEGY_ADAPTERS, + _safe_haven_cash_substitute_threshold_usd, + build_composer, + fetch_account_snapshot_token_with_metadata, ) -# Order pricing: limit order discount/premium relative to last price -LIMIT_SELL_DISCOUNT = 0.995 # sell limit at 0.5% below last -LIMIT_BUY_PREMIUM = 1.005 # buy limit at 0.5% above last -DEFAULT_LIMIT_BUY_PREMIUM_BY_SYMBOL = {"SOXL": 1.015, "TQQQ": 1.010} - -# Order monitoring: poll interval and max attempts for fill check -ORDER_POLL_INTERVAL_SEC = 1 -ORDER_POLL_MAX_ATTEMPTS = 8 - -# Token refresh: days before expiry to trigger refresh -TOKEN_REFRESH_THRESHOLD_DAYS = 30 -DEFAULT_SAFE_HAVEN_CASH_SUBSTITUTE_THRESHOLD_USD = 1000.0 -DEFAULT_MIN_ORDER_NOTIONAL_USD = 100.0 - -_FRACTIONAL_BUY_EXECUTION = fractional_buy_execution_enabled(STRATEGY_PROFILE) - - -def _profile_submit_order(t_ctx, symbol, **kwargs): - return submit_order( - t_ctx, - symbol, - allow_fractional_shares=_FRACTIONAL_BUY_EXECUTION, - quantity_step=FRACTIONAL_BUY_QUANTITY_STEP if _FRACTIONAL_BUY_EXECUTION else 1.0, - **kwargs, - ) - - -def _profile_estimate_max_purchase_quantity(t_ctx, symbol, **kwargs): - return estimate_max_purchase_quantity( - t_ctx, - symbol, - fractional_shares=_FRACTIONAL_BUY_EXECUTION, - **kwargs, - ) - -SEPARATOR = "━━━━━━━━━━━━━━━━━━" -COMPACT_ERROR_NOTIFICATION_MAX_CHARS = 3500 - - -def _load_limit_buy_premium_by_symbol(*env_names: str) -> dict[str, float]: - raw_value = "" - for env_name in env_names: - value = os.getenv(env_name) - if value and value.strip(): - raw_value = value.strip() - break - if not raw_value: - return dict(DEFAULT_LIMIT_BUY_PREMIUM_BY_SYMBOL) - try: - payload = json.loads(raw_value) - except json.JSONDecodeError as exc: - raise ValueError(f"Invalid limit buy premium map JSON: {raw_value!r}") from exc - if not isinstance(payload, dict): - raise ValueError("Limit buy premium map must be a JSON object keyed by symbol.") - parsed: dict[str, float] = {} - for symbol, premium in payload.items(): - symbol_text = str(symbol or "").strip().upper() - if not symbol_text: - continue - premium_value = float(premium) - if premium_value <= 0.0: - raise ValueError(f"Limit buy premium for {symbol_text} must be positive.") - parsed[symbol_text] = premium_value - return parsed - - -LIMIT_BUY_PREMIUM_BY_SYMBOL = _load_limit_buy_premium_by_symbol( - "LONGBRIDGE_LIMIT_BUY_PREMIUM_BY_SYMBOL_JSON", - "LIMIT_BUY_PREMIUM_BY_SYMBOL_JSON", +from notifications_runtime import ( + _runtime_failure_reason, + _compact_error_notification, + _split_env_list, + _runtime_error_notification_targets, + _runtime_error_notification_message, + _send_runtime_error, + _notify_runtime_error, + _handle_route_runtime_error, + _route_with_runtime_error_fallback, ) - -def t(key, **kwargs): - return build_translator(NOTIFY_LANG)(key, **kwargs) - - -def _runtime_failure_reason(exc: Exception) -> str: - # Match only fixed platform errors; never publish arbitrary provider text. - if isinstance(exc, RuntimeError) and str(exc) == "validation_journal_read_failed": - return "validation_journal_read_failed" - if isinstance(exc, RuntimeError) and str(exc) == "validation_journal_unavailable": - return "validation_journal_unavailable" - if isinstance(exc, RuntimeError) and str(exc) == "validation_journal_empty": - return "validation_journal_empty" - if isinstance(exc, RuntimeError) and str(exc) == "validation_side_effect_enabled": - return "validation_side_effect_enabled" - if isinstance(exc, RuntimeError) and str(exc) == "validation_marker_write_blocked": - return "validation_marker_write_blocked" - if isinstance(exc, RuntimeError) and str(exc) == "validation_read_only_bootstrap_unavailable": - return "validation_read_only_bootstrap_unavailable" - if isinstance(exc, RuntimeError) and str(exc) in { - "durable execution command producer is paper-only and cannot be enabled live", - "durable execution command consumer is paper-only and cannot be enabled live", - "strategy risk state persistence is paper-only and cannot be enabled live", - }: - return "paper_live_conflict" - return "unknown" - - -def _compact_error_notification( - exc: Exception, - *, - title: str | None = None, - prefix: str = "", - phase: str = "strategy_cycle", -) -> str: - error_text = type(exc).__name__ - known_phases = {"market_hours", "runtime_configuration", "strategy_cycle", "reporting", "health_probe"} - safe_phase = phase if phase in known_phases else "strategy_cycle" - reason = _runtime_failure_reason(exc) - message = "\n".join(( - title or t("error_title"), - t("failure_phase", phase=t(f"failure_phase_{safe_phase}")), - t(f"failure_reason_{reason}"), - f"{prefix}{error_text}", - )) - if len(message) <= COMPACT_ERROR_NOTIFICATION_MAX_CHARS: - return message - return message[: COMPACT_ERROR_NOTIFICATION_MAX_CHARS - 1].rstrip() + "…" - - -def _split_env_list(value: str | None) -> tuple[str, ...]: - return tuple( - item.strip() - for item in str(value or "").replace(";", ",").split(",") - if item.strip() - ) - - -def _summarize_cycle_result_for_report(cycle_result, *, dry_run: bool) -> dict: - if cycle_result is None: - return { - "action_done": False, - "order_events_count": 0, - "orders_previewed_count": 0, - "orders_skipped_count": 0, - "notes_count": 0, - "dry_run_order_preview_available": False, - } - logs = tuple(getattr(cycle_result, "logs", ()) or ()) - skip_logs = tuple(getattr(cycle_result, "skip_logs", ()) or ()) - note_logs = tuple(getattr(cycle_result, "note_logs", ()) or ()) - dry_run_orders = tuple(getattr(cycle_result, "dry_run_orders", ()) or ()) - pending_orders = tuple(getattr(cycle_result, "pending_orders", ()) or ()) - quote_snapshots = tuple(getattr(cycle_result, "quote_snapshots", ()) or ()) - order_events_count = 0 if pending_orders else len(logs) - orders_previewed_count = len(dry_run_orders) if dry_run_orders else (order_events_count if dry_run else 0) - broker_submission_done = bool(getattr(cycle_result, "action_done", False)) - summary = { - "action_done": broker_submission_done and not pending_orders, - "broker_submission_done": broker_submission_done, - "execution_status": ( - "pending_reconciliation" - if pending_orders - else ("previewed" if dry_run and broker_submission_done else "no_action") - ), - "order_events_count": order_events_count, - "orders_pending_count": len(pending_orders), - "orders_previewed_count": orders_previewed_count, - "orders_skipped_count": len(skip_logs), - "notes_count": len(note_logs), - "dry_run_order_preview_available": bool(dry_run and orders_previewed_count > 0), - } - if dry_run_orders: - summary["orders_previewed"] = [dict(order) for order in dry_run_orders] - if pending_orders: - summary["orders_pending"] = [dict(order) for order in pending_orders] - if quote_snapshots: - summary["quote_snapshot"] = { - "quotes": [dict(snapshot) for snapshot in quote_snapshots], - } - execution = dict(getattr(cycle_result, "execution", {}) or {}) - account_snapshot = execution.get("heartbeat_account_snapshot") - if isinstance(account_snapshot, dict): - summary["heartbeat_account_snapshot"] = dict(account_snapshot) - if execution.get("execution_status") == "blocked": - summary["execution_status"] = "blocked" - summary["blocked_reason"] = ( - "durable_live_execution_command_binding_invalid" - if execution.get("blocked_reason") == "durable_live_execution_command_binding_invalid" - else "unknown" - ) - for field in ( - "signal_date", - "effective_date", - "execution_timing_contract", - "execution_calendar_source", - ): - if execution.get(field) not in (None, ""): - summary[field] = execution[field] - durable_command = execution.get("durable_execution_command") - if isinstance(durable_command, dict): - summary["durable_execution_command"] = dict(durable_command) - live_command = execution.get("durable_live_execution_command") - if isinstance(live_command, dict): - summary["durable_live_execution_command"] = dict(live_command) - account_identity = execution.get("account_identity") - if isinstance(account_identity, dict): - summary["account_identity"] = dict(account_identity) - _, summary["notification_digest"] = unified_notifications.cycle_facts(cycle_result, dry_run=dry_run) - return summary - - -def _build_notification_delivery_log_for_report( - *, - platform: str, - strategy_profile: str, - run_id: str, - dry_run: bool, - orders_previewed_count: int, - delivery_events: list[dict], -) -> dict: - events = [dict(event) for event in delivery_events if dict(event).get("delivery_status") == "sent"] - if not dry_run or orders_previewed_count <= 0 or not events: - return {} - return { - "notification_schema_version": "hk_live_enablement_notification.v1", - "notification_event_type": "hk_snapshot_live_enablement_dry_run", - "notification_correlation_id": str(run_id or ""), - "locales": ["en", "zh-Hans"], - "profile": str(strategy_profile or ""), - "platform": str(platform or ""), - "validation_status": "passed", - "orders_previewed": int(orders_previewed_count), - "delivery_events": events, - "notification_contains_profile": True, - "notification_contains_platform": True, - "notification_contains_validation_status": True, - "notification_contains_order_preview_summary": True, - "notification_redacts_sensitive_fields": True, - "redaction_policy": "raw notification text is not persisted; only sha256 and length are recorded", - } - - -def _build_notification_delivery_summary(delivery_events: list[dict]) -> dict: - safe_fields = ( - "sink", - "delivery_status", - "transport_acknowledged", - "error_type", - "compact_text_sha256", - "compact_text_length", - ) - events = [ - {key: event[key] for key in safe_fields if key in event} - for event in (dict(item) for item in delivery_events) - ] - if not events: - return {} - sent_count = sum(event.get("delivery_status") == "sent" for event in events) - failed_count = sum(event.get("delivery_status") == "failed" for event in events) - return { - "attempted_count": len(events), - "sent_count": sent_count, - "failed_count": failed_count, - "all_acknowledged": failed_count == 0 and sent_count == len(events), - "delivery_events": events, - } - - -signal_text = build_signal_text(t) -strategy_display_name = build_strategy_display_name(t)( - STRATEGY_PROFILE, - fallback_name=STRATEGY_DISPLAY_NAME, - metadata=RUNTIME_SETTINGS.strategy_metadata, - locale=NOTIFY_LANG, +from strategy_plugins_runtime import ( + build_strategy_plugin_alert_messages, + build_strategy_plugin_alert_state_settings, + build_strategy_plugin_alert_context_label, + publish_strategy_plugin_alerts, ) - -def log_position_snapshot(message): - print(f"[{ACCOUNT_REGION}] {message}", flush=True) - - -def log_runtime_warning(message): - print(f"[{ACCOUNT_REGION}] [warning] {message}", flush=True) - - -BROKER_ADAPTERS = build_runtime_broker_adapters( - strategy_symbols=tuple(MANAGED_SYMBOLS), - account_hash=ACCOUNT_PREFIX or ACCOUNT_REGION or "longbridge", - fetch_last_price_fn=fetch_last_price, - fetch_strategy_account_state_fn=lambda quote_context, trade_context: fetch_strategy_account_state( - quote_context, - trade_context, - list(MANAGED_SYMBOLS), - cash_currency=TRADING_CURRENCY, - position_log_fn=( - log_position_snapshot - if getattr(RUNTIME_SETTINGS, "debug_position_snapshot", False) - else None - ), - warning_log_fn=log_runtime_warning, - ), - submit_order_fn=_profile_submit_order, - symbol_suffix=SYMBOL_SUFFIX, - currency=TRADING_CURRENCY, - cash_only_execution=CASH_ONLY_EXECUTION, -) -STRATEGY_ADAPTERS = build_runtime_strategy_adapters( - strategy_runtime=STRATEGY_RUNTIME, - strategy_profile=STRATEGY_PROFILE, - strategy_runtime_config=STRATEGY_RUNTIME_CONFIG, - available_inputs=AVAILABLE_INPUTS, - benchmark_symbol=BENCHMARK_SYMBOL, - signal_text_fn=signal_text, - translator=t, - broker_adapters=BROKER_ADAPTERS, - calculate_rotation_indicators_fn=calculate_rotation_indicators, - build_strategy_evaluation_inputs_fn=build_strategy_evaluation_inputs, - map_strategy_decision_to_plan_fn=map_strategy_decision_to_plan, - execution_policy={ - "reserved_cash_floor_usd": getattr(RUNTIME_SETTINGS, "reserved_cash_floor_usd", 0.0), - "reserved_cash_ratio": getattr(RUNTIME_SETTINGS, "reserved_cash_ratio", 0.0), - "cash_only_execution": CASH_ONLY_EXECUTION, - }, - execution_materials=getattr(STRATEGY_RUNTIME, "execution_materials", {}), - build_strategy_plugin_report_payload_fn=build_strategy_plugin_report_payload, - load_configured_strategy_plugin_signals_fn=load_configured_strategy_plugin_signals, - parse_strategy_plugin_mounts_fn=parse_strategy_plugin_mounts, +from http_routes import ( + app, + READ_ONLY_BROKER_RECONCILIATION_COLLECTOR, + _summarize_cycle_result_for_report, + _min_order_notional_usd, + run_strategy, + _bounded_probe_error_code, + run_probe, + _paper_command_consumer_session_date, + _paper_command_consumer_runtime_is_isolated, + run_paper_execution_command_consumer, + run_broker_reconciliation, + _open_account_history_store, + run_account_snapshot, + handle_trigger, + handle_backfill, + handle_dry_run, + handle_probe, + handle_paper_execution_command_consumer, + handle_broker_reconciliation, + handle_account_snapshot, + handle_v7_paper_application_loaded, + _v7_application_candidate_bound, + handle_monitor_dispatch, + request_method, + health, ) - -def _safe_haven_cash_substitute_threshold_usd() -> float: - return float( - getattr( - RUNTIME_SETTINGS, - "safe_haven_cash_substitute_threshold_usd", - DEFAULT_SAFE_HAVEN_CASH_SUBSTITUTE_THRESHOLD_USD, - ) - ) - - -def _min_order_notional_usd() -> float: - return float(getattr(RUNTIME_SETTINGS, "min_order_notional_usd", DEFAULT_MIN_ORDER_NOTIONAL_USD)) - - -def build_composer(*, dry_run_only_override: bool | None = None): - return build_runtime_composer( - project_id=PROJECT_ID, - secret_name=SECRET_NAME, - token_refresh_threshold_days=TOKEN_REFRESH_THRESHOLD_DAYS, - account_prefix=ACCOUNT_PREFIX, - account_region=ACCOUNT_REGION, - strategy_profile=STRATEGY_PROFILE, - strategy_display_name=STRATEGY_DISPLAY_NAME, - strategy_display_name_localized=strategy_display_name, - strategy_domain=RUNTIME_SETTINGS.strategy_domain, - notify_lang=NOTIFY_LANG, - tg_token=TG_TOKEN, - tg_chat_id=TG_CHAT_ID, - notification_channel=_NOTIFICATION_CHANNEL, - webhook_url=_NOTIFICATION_WEBHOOK_URL, - managed_symbols=tuple(MANAGED_SYMBOLS), - benchmark_symbol=BENCHMARK_SYMBOL, - signal_effective_after_trading_days=SIGNAL_EFFECTIVE_AFTER_TRADING_DAYS, - separator=SEPARATOR, - limit_sell_discount=LIMIT_SELL_DISCOUNT, - limit_buy_premium=LIMIT_BUY_PREMIUM, - limit_buy_premium_by_symbol=LIMIT_BUY_PREMIUM_BY_SYMBOL, - order_poll_interval_sec=ORDER_POLL_INTERVAL_SEC, - order_poll_max_attempts=ORDER_POLL_MAX_ATTEMPTS, - safe_haven_cash_substitute_threshold_usd=_safe_haven_cash_substitute_threshold_usd(), - min_order_notional_usd=_min_order_notional_usd(), - market=MARKET, - symbol_suffix=SYMBOL_SUFFIX, - trading_currency=TRADING_CURRENCY, - dry_run_only=RUNTIME_SETTINGS.dry_run_only, - dry_run_only_override=dry_run_only_override, - broker_adapters=BROKER_ADAPTERS, - strategy_adapters=STRATEGY_ADAPTERS, - estimate_max_purchase_quantity_fn=_profile_estimate_max_purchase_quantity, - fetch_order_status_fn=fetch_order_status, - fetch_token_from_secret_fn=fetch_token_from_secret, - fetch_token_with_metadata_fn=fetch_account_snapshot_token_with_metadata, - refresh_token_if_needed_fn=refresh_token_if_needed, - build_contexts_fn=build_contexts, - run_id_builder=build_run_id, - event_logger=emit_runtime_log, - report_builder=build_runtime_report_base, - report_persister=persist_runtime_report, - translator=t, - runtime_target=RUNTIME_SETTINGS.runtime_target, - env_reader=os.getenv, - sleeper=time.sleep, - printer=print, - extra_reporting_fields={ - "market": MARKET, - "market_calendar": MARKET_CALENDAR, - "market_timezone": MARKET_TIMEZONE, - "symbol_suffix": SYMBOL_SUFFIX, - "trading_currency": TRADING_CURRENCY, - }, - ) - - -def build_strategy_plugin_alert_messages(signals): - return STRATEGY_ADAPTERS.build_strategy_plugin_alert_messages(signals) - - -def _runtime_error_notification_targets() -> tuple[tuple[str, str], ...]: - targets: list[tuple[str, str]] = [] - if TG_TOKEN and TG_CHAT_ID: - targets.append((TG_TOKEN, TG_CHAT_ID)) - - seen: set[tuple[str, str]] = set() - unique_targets: list[tuple[str, str]] = [] - for target in targets: - if target in seen: - continue - seen.add(target) - unique_targets.append(target) - return tuple(unique_targets) - - -def _runtime_error_notification_message(exc: Exception, *, route_label: str) -> str: - title = "runtime_failure_title" - return "\n".join(( - t(title), - t("strategy_label", name=strategy_display_name or STRATEGY_PROFILE), - t("runtime_failure_context", context=ACCOUNT_REGION), - t("runtime_failure_result"), - t("runtime_failure_action"), - )) - - -def _send_runtime_error(exc: Exception, *, route_label: str) -> bool: - targets = _runtime_error_notification_targets() - if not targets: - print(t("runtime_notification_missing_target"), flush=True) - return False - message = _runtime_error_notification_message(exc, route_label=route_label) - outcomes = [] - for token, chat_id in targets: - try: - response = requests.post( - f"https://api.telegram.org/bot{token}/sendMessage", - json={"chat_id": chat_id, "text": message}, - timeout=10, - ) - status_code = int(getattr(response, "status_code", 200) or 200) - acknowledged = 200 <= status_code < 300 - load_payload = getattr(response, "json", None) - payload = load_payload() if acknowledged and callable(load_payload) else None - if isinstance(payload, dict) and payload.get("ok") is False: - acknowledged = False - outcomes.append(acknowledged) - except Exception: - print( - t("runtime_notification_delivery_failed"), - flush=True, - ) - outcomes.append(False) - return bool(outcomes) and all(outcomes) - - -def _notify_runtime_error(exc: Exception, *, route_label: str) -> bool: - return unified_notifications.dispatch_immediate( - lambda: _send_runtime_error(exc, route_label=route_label), kind="real_error", - incident=f"runtime_route:{route_label}:{_runtime_failure_reason(exc)}", - occurred_at=datetime.now(timezone.utc), run_id=build_run_id(), - ) if unified_notifications.enabled() else _send_runtime_error(exc, route_label=route_label) - - -def _handle_route_runtime_error(exc: Exception, *, route_label: str): - print(t("runtime_failure_log"), flush=True) - if validation_cycle_observation.current() is None: - _notify_runtime_error(exc, route_label=route_label) - return "Error", 500 - - -def _route_with_runtime_error_fallback(handler, *, success_body: str, route_label: str, **kwargs): - try: - if unified_notifications.enabled(): - with unified_notifications.bind_route(route_label): - result = handler(**kwargs) - else: - result = handler(**kwargs) - except Exception as exc: - return _handle_route_runtime_error(exc, route_label=route_label) - if result is False: - return "Error", 500 - if isinstance(result, tuple): - return result - return success_body, 200 - - -def build_strategy_plugin_alert_state_settings(): - return StrategyPluginAlertStateSettings.from_env( - project_id=PROJECT_ID, - ) - - -def build_strategy_plugin_alert_context_label() -> str: - return build_alert_context_label( - platform_id="longbridge", - strategy_profile=STRATEGY_PROFILE, - account_scope=ACCOUNT_REGION, - service_name=SECRET_NAME, - runtime_target=RUNTIME_SETTINGS.runtime_target, - ) - - -def publish_strategy_plugin_alerts(signals, *, report=None): - from notifications.digest_snapshot.digest_routing import immediate_plugin_signals - from notifications.digest_snapshot.digest_candidates import routine_observation_counts - if report is not None: - report.setdefault("summary", {})["routine_observation_counts"] = routine_observation_counts(signals) - result = dispatch_strategy_plugin_alerts( - immediate_plugin_signals(signals, mode="daily_digest"), - notification_settings=RUNTIME_SETTINGS, - translator=t, - strategy_label=STRATEGY_PROFILE, - context_label=build_strategy_plugin_alert_context_label(), - state_settings=build_strategy_plugin_alert_state_settings(), - log_message=print, - ) - if report is not None: - result.attach_to_report(report) - return result - - -def run_strategy(*, force_run: bool = False, validation_only: bool = False, validation_label: str = "backfill"): - if _v7_application_candidate_bound() and not validation_only: - print(f"[{datetime.now()}] V7 paper candidate is paused; skip all execution paths.", flush=True) - return True - if not validation_only and not force_run and not getattr(RUNTIME_SETTINGS, "runtime_target_enabled", True): - print(f"[{datetime.now()}] Runtime target disabled; skip strategy execution.", flush=True) - return True - composer = build_composer(dry_run_only_override=True if validation_only else None) - observation = validation_cycle_observation.current() if validation_only else None - if observation is not None: - composer = validation_cycle_observation.instrument_composer(composer, observation) - reporting_adapters = composer.build_reporting_adapters() - log_context, report = reporting_adapters.start_run() - notification_delivery_events: list[dict] = [] - try: - notification_adapters = composer.build_notification_adapters( - delivery_events=notification_delivery_events, - ) - except TypeError as exc: - if "delivery_events" not in str(exc): - raise - notification_adapters = composer.build_notification_adapters() - strategy_plugin_signals, strategy_plugin_error = composer.load_strategy_plugin_signals( - getattr(RUNTIME_SETTINGS, "strategy_plugin_mounts_json", None) - ) - composer.attach_strategy_plugin_report( - report, - signals=strategy_plugin_signals, - error=strategy_plugin_error, - ) - failure_phase = "market_hours" - try: - reporting_adapters.log_event( - log_context, - "strategy_cycle_started", - message="Starting strategy execution", - ) - print(composer.with_prefix(f"[{datetime.now()}] Starting strategy..."), flush=True) - - market_open = is_market_open_now( - calendar_name=MARKET_CALENDAR, - timezone_name=MARKET_TIMEZONE, - ) - if isinstance(market_open, tuple): - market_open, error = market_open - if error is not None: - reporting_adapters.log_event( - log_context, - "market_hours_check_failed", - message="Market hours check failed", - severity="WARNING", - error_message=str(error), - market=MARKET, - market_calendar=MARKET_CALENDAR, - market_timezone=MARKET_TIMEZONE, - ) - print(composer.with_prefix(f"Market hours check failed: {error}"), flush=True) - if not market_open and not force_run: - reporting_adapters.log_event( - log_context, - "outside_market_hours", - message="Outside market hours; skip execution", - market=MARKET, - market_calendar=MARKET_CALENDAR, - market_timezone=MARKET_TIMEZONE, - ) - finalize_runtime_report( - report, - status="skipped", - diagnostics={ - "skip_reason": "market_closed", - }, - ) - print(composer.with_prefix("Outside market hours; skip."), flush=True) - return True - if force_run and not market_open: - reporting_adapters.log_event( - log_context, - "market_hours_bypassed", - message=f"Market hours bypassed for {validation_label} execution", - market=MARKET, - market_calendar=MARKET_CALENDAR, - market_timezone=MARKET_TIMEZONE, - ) - print( - composer.with_prefix( - f"Market hours bypassed for {validation_label} verification; validation only, no orders will be submitted." - ), - flush=True, - ) - unsupported_reason = dca_execution_unsupported_reason(STRATEGY_PROFILE) - if unsupported_reason is not None: - reporting_adapters.log_event( - log_context, - "strategy_execution_unsupported", - message="Strategy requires fractional-share execution; skip", - skip_reason=unsupported_reason, - strategy_profile=STRATEGY_PROFILE, - market=MARKET, - market_calendar=MARKET_CALENDAR, - market_timezone=MARKET_TIMEZONE, - ) - finalize_runtime_report( - report, - status="skipped", - diagnostics={"skip_reason": unsupported_reason}, - ) - print( - composer.with_prefix( - f"Strategy {STRATEGY_PROFILE} requires fractional-share execution; skip." - ), - flush=True, - ) - return True - if not validation_only: - publish_strategy_plugin_alerts(strategy_plugin_signals, report=report) - failure_phase = "runtime_configuration" - try: - rebalance_runtime = composer.build_rebalance_runtime( - silent_cycle_notifications=validation_only, - notification_delivery_events=notification_delivery_events, - ) - except TypeError as exc: - if "notification_delivery_events" not in str(exc): - raise - rebalance_runtime = composer.build_rebalance_runtime( - silent_cycle_notifications=validation_only, - ) - if observation is not None: - validation_bootstrap = getattr(rebalance_runtime.bootstrap, "validation_cycle", None) - if not callable(validation_bootstrap): - raise RuntimeError("validation_read_only_bootstrap_unavailable") - rebalance_runtime = dataclasses.replace( - rebalance_runtime, bootstrap=validation_bootstrap, - notify_issue=lambda _title, _detail: None, - ) - rebalance_config = composer.build_rebalance_config( - strategy_plugin_signals=strategy_plugin_signals, - strategy_plugin_error=strategy_plugin_error, - notification_title_key=( - "dry_run_title" - if validation_only and validation_label == "dry_run" - else "" - ), - cash_only_execution=CASH_ONLY_EXECUTION, - live_execution_session_authorized=bool(market_open and not validation_only), - ) - if observation is not None: - validation_cycle_observation.require_matching_journal_source(rebalance_config, reporting_adapters) - observation.report_base_dir = getattr(reporting_adapters, "report_base_dir", None) - failure_phase = "strategy_cycle" - if observation is not None: - observation.cycle_entered = True - with unified_notifications.bind_report(report, validation_only=validation_only): - cycle_result = run_rebalance_cycle(runtime=rebalance_runtime, config=rebalance_config) - if observation is not None: - observation.cycle_returned = True - failure_phase = "reporting" - signal_snapshot = {} - if cycle_result is not None: - execution = dict(getattr(cycle_result, "execution", {}) or {}) - signal_snapshot = dict(execution.get("signal_snapshot") or {}) - execution_summary = _summarize_cycle_result_for_report( - cycle_result, - dry_run=bool(report.get("dry_run")), - ) - if "routine_observation_counts" in (report.get("summary") or {}): - execution_summary["routine_observation_counts"] = report["summary"]["routine_observation_counts"] - if not validation_only: - try: - unified_notifications.attach_cycle_event(execution_summary, cycle_result, report=report, config=rebalance_config) - except Exception: - print("notification_event_attachment_unconfirmed", flush=True) - unified_notifications.record_failure("notification_event_attachment_unconfirmed", report=report) - notification_delivery_log = _build_notification_delivery_log_for_report( - platform="longbridge", - strategy_profile=STRATEGY_PROFILE, - run_id=str(report.get("run_id") or ""), - dry_run=bool(report.get("dry_run")), - orders_previewed_count=int(execution_summary.get("orders_previewed_count") or 0), - delivery_events=notification_delivery_events, - ) - if notification_delivery_log: - execution_summary["notification_delivery_log"] = notification_delivery_log - if unified_notifications.enabled(): - notification_delivery_events.extend( - {"sink": "notification_store", "delivery_status": "failed", "transport_acknowledged": False, - "error_type": code} for code in report.get("notification_unified_errors", ()) - ) - notification_delivery_summary = _build_notification_delivery_summary( - notification_delivery_events - ) - if notification_delivery_summary: - execution_summary["notification_delivery_summary"] = notification_delivery_summary - if signal_snapshot: - reporting_adapters.log_event( - log_context, - "strategy_signal_snapshot", - message="Strategy signal snapshot", - **signal_snapshot, - ) - try: - attach_cycle_execution_receipt(report, cycle_result) - except ValueError: - # Test/legacy runtime reports without a release attestation remain - # evidence-missing; receipt metadata must not change a completed - # strategy cycle into a failure. - pass - finalize_runtime_report( - report, - status="ok", - summary=execution_summary, - diagnostics={"signal_snapshot": signal_snapshot} if signal_snapshot else None, - ) - reporting_adapters.log_event( - log_context, - "strategy_cycle_completed", - message="Strategy execution completed", - ) - return True - - except Exception as exc: - failure_reason = _runtime_failure_reason(exc) - append_runtime_report_error( - report, - stage="strategy_cycle", - message="strategy_cycle_failed", - error_type=type(exc).__name__, - ) - reporting_adapters.log_event( - log_context, - "strategy_cycle_failed", - message="Strategy execution failed", - severity="ERROR", - error_type=type(exc).__name__, - error_message="strategy_cycle_failed", - failure_phase=failure_phase, - failure_reason=failure_reason, - ) - if observation is None and not (validation_only and _v7_application_candidate_bound()): - try: - def publish_failure(): - return notification_adapters.publish_cycle_notification( - detailed_text=_compact_error_notification(exc, phase=failure_phase), - compact_text=_compact_error_notification(exc, phase=failure_phase), - ) - unified_notifications.dispatch_immediate(publish_failure, kind="real_error", - incident=f"runtime_failure:{failure_phase}:{failure_reason}", report=report, actually_ran=True) - except Exception as notification_exc: - notification_delivery_events.append( - { - "sink": "telegram", - "delivery_status": "failed", - "transport_acknowledged": False, - "error_type": type(notification_exc).__name__, - } - ) - reporting_adapters.log_event( - log_context, - "strategy_error_notification_failed", - message="Strategy error notification failed", - severity="ERROR", - error_type=type(notification_exc).__name__, - ) - error_summary = {} - notification_delivery_summary = _build_notification_delivery_summary( - notification_delivery_events - ) - if notification_delivery_summary: - error_summary["notification_delivery_summary"] = ( - notification_delivery_summary - ) - finalize_runtime_report( - report, - status="error", - summary=error_summary or None, - diagnostics={"failure_phase": failure_phase, "failure_reason": failure_reason}, - ) - return False - finally: - if "execution_receipt" not in report: - try: - attach_terminal_fallback_execution_receipt(report) - except ValueError: - # Legacy targets remain evidence-missing; receipt metadata - # cannot affect strategy execution or error handling. - pass - if observation is not None: - status = str(report.get("status") or "").strip().lower() - observation.report_status = status if status in {"ok", "error", "skipped"} else "unknown" - observation.execution_receipt_present = isinstance(report.get("execution_receipt"), dict) - if observation is None and not (validation_only and _v7_application_candidate_bound()): - try: - report_path = reporting_adapters.persist_execution_report(report) - print(f"execution_report {report_path}", flush=True) - if not validation_only and report.get("status") in {"ok", "completed", "success"}: - if not unified_notifications.report_archive_confirmed( - report, report_path, reporting_adapters.report_gcs_prefix_uri - ): - raise ValueError("notification_report_archive_unconfirmed") - if unified_notifications.persist_archived_report(report, env=os.environ) is not True: - raise ValueError("notification_archive_candidate_unconfirmed") - if observation is None and unified_notifications.enabled(): - unified_notifications.recover_runtime(report=report) - except Exception as persist_exc: - print(f"failed to persist execution report: {type(persist_exc).__name__}", flush=True) - if not validation_only and report.get("status") in {"ok", "completed", "success"}: - unified_notifications._storage_failure( - lambda title, detail: rebalance_runtime.notifications.send_text(f"{title}\n{detail}") - ) - - -def _bounded_probe_error_code(exc: Exception) -> int | None: - try: - code = getattr(exc, "code", None) - except Exception: - return None - if type(code) is not int or not -(2**31) <= code <= 2**31 - 1: - return None - return code - - -def run_probe(*, response_body: str = "Probe OK"): - if _v7_application_candidate_bound(): - return response_body, 200 - history_scope = str(ACCOUNT_REGION or "").strip() - record_account_history = ( - history_scope in {"HK", "SG"} - and str(os.getenv("ACCOUNT_HISTORY_RECORDING_ENABLED") or "").strip() == "true" - ) - history_capture = begin_natural_cycle_history() if record_account_history else None - composer = None - reporting_adapters = None - log_context = None - report = None - probe_step = "settings" - try: - composer = build_composer(dry_run_only_override=True) - reporting_adapters = composer.build_reporting_adapters() - log_context, report = reporting_adapters.start_run() - reporting_adapters.log_event( - log_context, - "health_probe_received", - message="Received health probe request", - execution_window="probe", - ) - probe_step = "context" - runtime = composer.build_rebalance_runtime(silent_cycle_notifications=True) - if record_account_history: - probe_step = "context" - quote_context, trade_context, source_binding = composer.build_account_snapshot_broker_contexts() - probe_step = "balance" - read_cycle_account_balance(lambda: trade_context.account_balance(currency="USD")) - # Keep known balances even when the optional stock read fails. The - # portfolio below consumes the same response/error, without a retry. - stock_reader = getattr(trade_context, "stock_positions", None) - if callable(stock_reader): - try: - read_cycle_stock_positions(stock_reader) - except Exception: - pass - observation = cycle_history_observation() - if observation is None: - raise RuntimeError("account history observation incomplete") - probe_step = "archive" - result = record_projected_daily_account( - os.environ, - account_scope=history_scope, - source_binding=source_binding, - balances=observation["projection"]["broker_reported_balances"], - cash=observation["projection"]["cash"], - started=observation["started"], - finished=observation["finished"], - open_store=_open_account_history_store, - financing=observation["projection"].get("financing"), - positions=observation["projection"].get("broker_reported_positions"), - ) - print(f"account_history status={result.status} category={result.category or '-'}", flush=True) - if result.status != "recorded": - raise RuntimeError("account history write was not confirmed") - probe_step = "indicators" - _indicators = composer.strategy_adapters.calculate_strategy_indicators(quote_context) - else: - probe_step = "bootstrap" - quote_context, trade_context, _indicators = runtime.bootstrap() - probe_step = "portfolio_snapshot" - snapshot = runtime.portfolio_port_factory( - quote_context, - trade_context, - ).get_portfolio_snapshot() - positions = tuple(getattr(snapshot, "positions", ()) or ()) - buying_power = float(getattr(snapshot, "buying_power", 0.0) or 0.0) - total_equity = float(getattr(snapshot, "total_equity", 0.0) or 0.0) - finalize_runtime_report( - report, - status="ok", - summary={ - "buying_power": buying_power, - "total_equity": total_equity, - "positions_count": len(positions), - }, - ) - reporting_adapters.log_event( - log_context, - "health_probe_completed", - message="Health probe completed", - execution_window="probe", - buying_power=buying_power, - total_equity=total_equity, - positions_count=len(positions), - ) - return response_body, 200 - except Exception as exc: - error_code = _bounded_probe_error_code(exc) - error_diagnostics = { - "probe_step": probe_step, - "error_code_status": "known" if error_code is not None else "unknown", - } - if error_code is not None: - error_diagnostics["error_code"] = error_code - if report is not None: - append_runtime_report_error( - report, - stage="health_probe", - message="health_probe_failed", - error_type=type(exc).__name__, - **error_diagnostics, - ) - finalize_runtime_report(report, status="error") - if reporting_adapters is not None and log_context is not None: - reporting_adapters.log_event( - log_context, - "health_probe_failed", - message="Health probe failed", - severity="ERROR", - execution_window="probe", - error_type=type(exc).__name__, - error_message="health_probe_failed", - **error_diagnostics, - ) - err = _compact_error_notification( - exc, title=t("health_probe_title"), prefix=t("health_probe_error_prefix"), phase="health_probe", - ) - if composer is not None: - def publish_probe_failure(): - return composer.build_notification_adapters().publish_cycle_notification( - detailed_text=err, - compact_text=_compact_error_notification( - exc, title=t("health_probe_title"), prefix=t("health_probe_error_prefix"), phase="health_probe", - ), - ) - unified_notifications.dispatch_immediate(publish_probe_failure, kind="real_error", - incident="runtime_failure:health_probe", report=report) - else: - print(err, flush=True) - return "Error", 500 - finally: - if history_capture is not None: - end_natural_cycle_history(history_capture) - try: - if reporting_adapters is not None and report is not None: - report_path = reporting_adapters.persist_execution_report(report) - print(f"execution_report {report_path}", flush=True) - if unified_notifications.enabled(): - unified_notifications.recover_runtime(report=report, probe=True) - except Exception as persist_exc: - print(f"failed to persist execution report: {type(persist_exc).__name__}", flush=True) - - -def _paper_command_consumer_session_date() -> str: - """Return the exchange-local session date used to claim due commands.""" - try: - import pytz - - return datetime.now(pytz.timezone(MARKET_TIMEZONE)).date().isoformat() - except Exception: - return datetime.now().date().isoformat() - - -def _paper_command_consumer_runtime_is_isolated() -> bool: - runtime_target = getattr(RUNTIME_SETTINGS, "runtime_target", None) - return bool( - getattr(RUNTIME_SETTINGS, "dry_run_only", False) - and runtime_target is not None - and str(getattr(runtime_target, "execution_mode", "") or "").lower() == "paper" - ) - - -def run_paper_execution_command_consumer() -> bool: - """Run an explicitly invoked, read-only paper command consumer. - - This handler never constructs an execution port or calls the LongBridge - order API. It is deliberately separate from ``/run`` so a normal cycle - cannot start consuming delayed commands by accident. - """ - if _v7_application_candidate_bound(): - return True - if not _paper_command_consumer_runtime_is_isolated(): - raise RuntimeError( - "paper command consumer requires RUNTIME_TARGET_JSON.execution_mode=paper " - "and LONGBRIDGE_DRY_RUN_ONLY=true" - ) - if not resolve_paper_execution_command_consumer_enabled( - env_reader=os.getenv, - dry_run_only=bool(RUNTIME_SETTINGS.dry_run_only), - ): - raise RuntimeError("paper command consumer is not enabled") - - composer = build_composer() - config = composer.build_rebalance_config() - reporting_adapters = composer.build_reporting_adapters() - log_context, report = reporting_adapters.start_run() - try: - reporting_adapters.log_event( - log_context, - "paper_execution_command_consumer_started", - message="Starting read-only paper execution command consumer", - ) - quote_context, trade_context = composer.build_read_only_broker_contexts() - portfolio = composer.broker_adapters.build_portfolio_port( - quote_context, - trade_context, - ).get_portfolio_snapshot() - result = consume_due_paper_execution_commands( - store=config.execution_command_store, - as_of_session=_paper_command_consumer_session_date(), - claimant=str(os.getenv("K_SERVICE") or "longbridge-paper-command-consumer"), - portfolio=portfolio, - market_data_port=composer.broker_adapters.build_market_data_port(quote_context), - runtime_release_receipt=config.runtime_release_receipt, - expected_strategy_release=config.expected_strategy_release, - expected_command_binding={ - "platform": "longbridge", - "account_scope": str(config.execution_state_account_scope or "unknown"), - "strategy_profile": str(config.strategy_profile or "unknown"), - }, - ) - report_status = "ok" if result.get("status") == "ok" else "skipped" - finalize_runtime_report( - report, - status=report_status, - summary={"paper_execution_command_consumer": result}, - ) - reporting_adapters.log_event( - log_context, - "paper_execution_command_consumer_completed", - message="Paper execution command consumer completed", - result_status=result.get("status"), - commands_count=len(tuple(result.get("commands") or ())), - ) - return True - except Exception as exc: - append_runtime_report_error( - report, - stage="paper_execution_command_consumer", - message=str(exc), - error_type=type(exc).__name__, - ) - finalize_runtime_report(report, status="error") - reporting_adapters.log_event( - log_context, - "paper_execution_command_consumer_failed", - message="Paper execution command consumer failed", - severity="ERROR", - error_type=type(exc).__name__, - error_message=str(exc), - ) - return False - finally: - try: - report_path = reporting_adapters.persist_execution_report(report) - print(f"execution_report {report_path}", flush=True) - except Exception as persist_exc: - print(f"failed to persist execution report: {persist_exc}", flush=True) - - -def run_broker_reconciliation(): - """Run the explicitly enabled, order-port-free reconciliation boundary.""" - - return run_read_only_broker_reconciliation( - enabled=reconciliation_enabled(os.getenv), - account_scope=getattr(RUNTIME_SETTINGS, "account_region", None), - runtime_target=getattr(RUNTIME_SETTINGS, "runtime_target", None), - strategy_profile=STRATEGY_PROFILE, - project_id=PROJECT_ID, - build_read_only_contexts=lambda: build_composer().build_read_only_broker_contexts(), - collect_evidence=READ_ONLY_BROKER_RECONCILIATION_COLLECTOR, - ) - - -def fetch_account_snapshot_token_with_metadata(project_id, secret_name): - """Read one configured secret-store response. A metadata failure is not reread.""" - from types import SimpleNamespace - - from quant_platform_kit.cloud import get_secret_store - - store = get_secret_store() - read_metadata = getattr(store, "get_secret_with_metadata", None) - if callable(read_metadata): - return read_metadata(secret_name, project_id=project_id) - return SimpleNamespace( - value=store.get_secret(secret_name, project_id=project_id), - version_name=None, - ) - - -def _open_account_history_store(project_id): - from quant_platform_kit.cloud import get_object_store - - return get_object_store(project_id=project_id) - - -def run_account_snapshot(): - """Read bounded account facts through the order-port-free broker contexts.""" - - return run_read_only_account_snapshot( - enabled=account_snapshot_enabled(os.getenv), - account_scope=getattr(RUNTIME_SETTINGS, "account_region", None), - build_read_only_contexts=lambda: build_composer().build_account_snapshot_broker_contexts(), - collect_evidence=READ_ONLY_BROKER_RECONCILIATION_COLLECTOR, - ) - - -@app.route("/run", methods=["POST"]) -def handle_trigger(): - """Entrypoint for Cloud Run / scheduler: run strategy and return 200.""" - if request_method() != "POST": - return "Method Not Allowed", 405 - return _route_with_runtime_error_fallback( - run_strategy, - success_body="OK", - route_label="POST /run", - ) - - -@app.route("/backfill", methods=["POST", "GET"]) -def handle_backfill(): - """Manual backfill entrypoint for verification-only execution.""" - def run_validation(): - return _route_with_runtime_error_fallback( - run_strategy, - force_run=True, - validation_only=True, - success_body="OK", - route_label="POST /backfill", - ) - - if str(ACCOUNT_REGION or "").strip().upper() != "PAPER": - return run_validation() - headers = getattr(request, "headers", {}) - with validation_cycle_observation.observe(headers.get("X-Cloud-Trace-Context"), PROJECT_ID): - return run_validation() - - -@app.route("/dry-run", methods=["POST", "GET"]) -def handle_dry_run(): - """Strategy dry-run entrypoint.""" - def run_validation(): - return _route_with_runtime_error_fallback( - run_strategy, - force_run=True, - validation_only=True, - validation_label="dry_run", - success_body="Dry Run OK", - route_label="POST /dry-run", - ) - - if str(ACCOUNT_REGION or "").strip().upper() != "PAPER": - return run_validation() - headers = getattr(request, "headers", {}) - with validation_cycle_observation.observe(headers.get("X-Cloud-Trace-Context"), PROJECT_ID): - return run_validation() - - -@app.route("/probe", methods=["POST"]) -def handle_probe(): - """Post-open broker/account health probe; notify only on failure.""" - if request_method() != "POST": - return "Method Not Allowed", 405 - return _route_with_runtime_error_fallback( - run_probe, - success_body="Probe OK", - route_label="POST /probe", - ) - - -@app.route("/paper-command-consumer", methods=["POST"]) -def handle_paper_execution_command_consumer(): - """Explicit paper-only command verification endpoint; never scheduler-routed.""" - return _route_with_runtime_error_fallback( - run_paper_execution_command_consumer, - success_body="Paper command consumer OK", - route_label="POST /paper-command-consumer", - ) - - -@app.route("/reconcile", methods=["POST"]) -def handle_broker_reconciliation(): - """Return only redacted QPK E3 evidence; never invoke normal execution.""" - - if request_method() != "POST": - return "Method Not Allowed", 405 - payload, status = run_broker_reconciliation() - return json.dumps(payload, ensure_ascii=True), status, {"Content-Type": "application/json"} - - -@app.route("/account-snapshot", methods=["GET"]) -def handle_account_snapshot(): - """Return a non-cacheable, read-only and authority-free account snapshot.""" - - if request_method() != "GET": - return "Method Not Allowed", 405 - payload, status = run_account_snapshot() - return json.dumps(payload, ensure_ascii=True), status, { - "Content-Type": "application/json", - "Cache-Control": "no-store", - } - - -@app.route("/v7-paper-application-loaded", methods=["GET"]) -def handle_v7_paper_application_loaded(): - """Read back the disabled V7 process identity without broker access.""" - - if request_method() != "GET": - return "Method Not Allowed", 405 - from application.v7_paper_application import ( - V7PaperApplicationError, - build_v7_loaded_version_readback, - load_v7_paper_application_binding, - ) - try: - binding = load_v7_paper_application_binding(os.environ) - if binding is None: - return json.dumps( - {"status": "unavailable", "reason": "application_binding_missing"}, - ensure_ascii=True, - ), 409, {"Content-Type": "application/json", "Cache-Control": "no-store"} - payload = build_v7_loaded_version_readback( - binding, - runtime_settings=RUNTIME_SETTINGS, - strategy_runtime=STRATEGY_RUNTIME, - runtime_env=os.environ, - ) - except V7PaperApplicationError: - return json.dumps( - {"status": "uncertain", "reason": "loaded_version_unavailable"}, - ensure_ascii=True, - ), 409, {"Content-Type": "application/json", "Cache-Control": "no-store"} - return json.dumps( - {"status": "applied_paused", "readback": payload}, - ensure_ascii=True, - ), 200, {"Content-Type": "application/json", "Cache-Control": "no-store"} - - -def _v7_application_candidate_bound() -> bool: - """Keep the exact V7 binding paused even when a route bypasses the flag.""" - - if STRATEGY_PROFILE != V7_LONGTERM_COMPOUNDING_PROFILE: - return False - from application.v7_paper_application import load_v7_paper_application_binding - - return load_v7_paper_application_binding(os.environ) is not None - - -@app.route("/monitor-dispatch", methods=["POST", "GET"]) -def handle_monitor_dispatch(): - if request_method() == "GET": - return "Monitor Dispatch OK - use POST to dispatch due monitor checks", 200 - try: - result = dispatch_due_monitors(_load_monitor_dispatch_targets()) - except Exception as exc: - return _handle_route_runtime_error(exc, route_label="POST /monitor-dispatch") - return json.dumps(result, ensure_ascii=False), 200, {"Content-Type": "application/json"} - - -def request_method() -> str: - try: - return request.method - except Exception: - return "GET" - - -@app.route("/healthz", methods=["GET"]) -@app.route("/health", methods=["GET"]) -def health(): - critical_errors: list[str] = [] - for module_path in ( - "application.runtime_composer", - "application.runtime_reporting_adapters", - "application.runtime_strategy_adapters", - "runtime_config_support", - "runtime_execution_policy", - ): - try: - importlib.import_module(module_path) - except Exception as exc: - critical_errors.append(f"{module_path}: {type(exc).__name__}: {exc}") - if critical_errors: - return json.dumps({"status": "unhealthy", "errors": critical_errors}, ensure_ascii=False), 500 - return "OK", 200 - +__all__ = [ + 'FRACTIONAL_BUY_QUANTITY_STEP', + 'Flask', + 'StartupConfigReceipt', + 'StrategyPluginAlertStateSettings', + 'V7_LONGTERM_COMPOUNDING_PROFILE', + 'account_snapshot_enabled', + 'append_runtime_report_error', + 'attach_cycle_execution_receipt', + 'attach_terminal_fallback_execution_receipt', + 'begin_natural_cycle_history', + 'build_alert_context_label', + 'build_contexts', + 'build_run_id', + 'build_runtime_broker_adapters', + 'build_runtime_composer', + 'build_runtime_report_base', + 'build_runtime_strategy_adapters', + 'build_signal_text', + 'build_strategy_display_name', + 'build_strategy_evaluation_inputs', + 'build_strategy_plugin_report_payload', + 'build_translator', + 'calculate_rotation_indicators', + 'collect_read_only_reconciliation_observations', + 'consume_due_paper_execution_commands', + 'cycle_history_observation', + 'dataclasses', + 'datetime', + 'dca_execution_unsupported_reason', + 'dispatch_due_monitors', + 'dispatch_strategy_plugin_alerts', + 'emit_runtime_log', + 'end_natural_cycle_history', + 'estimate_max_purchase_quantity', + 'fetch_last_price', + 'fetch_order_status', + 'fetch_strategy_account_state', + 'fetch_token_from_secret', + 'finalize_runtime_report', + 'fractional_buy_execution_enabled', + 'google', + 'importlib', + 'is_market_open_now', + 'json', + 'load_configured_strategy_plugin_signals', + 'load_monitor_targets', + 'load_platform_runtime_settings', + 'load_strategy_runtime', + 'map_strategy_decision_to_plan', + 'normalize_monitor_dispatch_targets', + 'normalize_plugin_mounts_strategy', + 'os', + 'overlay_env', + 'parse_strategy_plugin_mounts', + 'persist_runtime_report', + 'read_cycle_account_balance', + 'read_cycle_stock_positions', + 'reconciliation_enabled', + 'record_projected_daily_account', + 'refresh_token_if_needed', + 'request', + 'requests', + 'resolve_cycle_channel_and_url', + 'resolve_paper_execution_command_consumer_enabled', + 'run_read_only_account_snapshot', + 'run_read_only_broker_reconciliation', + 'run_rebalance_cycle', + 'submit_order', + 'time', + 'timezone', + 'unified_notifications', + 'validation_cycle_observation', + 'get_project_id', + 'RUNTIME_SETTINGS', + 'PROJECT_ID', + 'SECRET_NAME', + 'ACCOUNT_PREFIX', + 'STRATEGY_PROFILE', + 'STRATEGY_DISPLAY_NAME', + '_normalize_plugin_mounts_strategy', + '_adopt_startup_config', + 'MONITOR_DISPATCH_ENV_OVERRIDES', + 'STARTUP_CONFIG_RECEIPT', + '_load_monitor_dispatch_targets', + 'ACCOUNT_REGION', + 'MARKET', + 'MARKET_CALENDAR', + 'MARKET_TIMEZONE', + 'SYMBOL_SUFFIX', + 'TRADING_CURRENCY', + 'NOTIFY_LANG', + 'TG_TOKEN', + 'TG_CHAT_ID', + '_NOTIFICATION_CHANNEL', + '_NOTIFICATION_WEBHOOK_URL', + 'CASH_ONLY_EXECUTION', + 'STRATEGY_RUNTIME', + 'STRATEGY_RUNTIME_CONFIG', + 'MANAGED_SYMBOLS', + 'AVAILABLE_INPUTS', + 'BENCHMARK_SYMBOL', + '_emit_v7_loaded_version_log', + 'SIGNAL_EFFECTIVE_AFTER_TRADING_DAYS', + 'LIMIT_SELL_DISCOUNT', + 'LIMIT_BUY_PREMIUM', + 'DEFAULT_LIMIT_BUY_PREMIUM_BY_SYMBOL', + 'ORDER_POLL_INTERVAL_SEC', + 'ORDER_POLL_MAX_ATTEMPTS', + 'TOKEN_REFRESH_THRESHOLD_DAYS', + 'DEFAULT_SAFE_HAVEN_CASH_SUBSTITUTE_THRESHOLD_USD', + 'DEFAULT_MIN_ORDER_NOTIONAL_USD', + '_FRACTIONAL_BUY_EXECUTION', + '_profile_submit_order', + '_profile_estimate_max_purchase_quantity', + 'SEPARATOR', + 'COMPACT_ERROR_NOTIFICATION_MAX_CHARS', + '_load_limit_buy_premium_by_symbol', + 'LIMIT_BUY_PREMIUM_BY_SYMBOL', + 't', + '_build_notification_delivery_log_for_report', + '_build_notification_delivery_summary', + 'signal_text', + 'strategy_display_name', + 'log_position_snapshot', + 'log_runtime_warning', + 'BROKER_ADAPTERS', + 'STRATEGY_ADAPTERS', + '_safe_haven_cash_substitute_threshold_usd', + 'build_composer', + 'fetch_account_snapshot_token_with_metadata', + '_runtime_failure_reason', + '_compact_error_notification', + '_split_env_list', + '_runtime_error_notification_targets', + '_runtime_error_notification_message', + '_send_runtime_error', + '_notify_runtime_error', + '_handle_route_runtime_error', + '_route_with_runtime_error_fallback', + 'build_strategy_plugin_alert_messages', + 'build_strategy_plugin_alert_state_settings', + 'build_strategy_plugin_alert_context_label', + 'publish_strategy_plugin_alerts', + 'app', + 'READ_ONLY_BROKER_RECONCILIATION_COLLECTOR', + '_summarize_cycle_result_for_report', + '_min_order_notional_usd', + 'run_strategy', + '_bounded_probe_error_code', + 'run_probe', + '_paper_command_consumer_session_date', + '_paper_command_consumer_runtime_is_isolated', + 'run_paper_execution_command_consumer', + 'run_broker_reconciliation', + '_open_account_history_store', + 'run_account_snapshot', + 'handle_trigger', + 'handle_backfill', + 'handle_dry_run', + 'handle_probe', + 'handle_paper_execution_command_consumer', + 'handle_broker_reconciliation', + 'handle_account_snapshot', + 'handle_v7_paper_application_loaded', + '_v7_application_candidate_bound', + 'handle_monitor_dispatch', + 'request_method', + 'health', +] if __name__ == "__main__": - app.run(host='0.0.0.0', port=int(os.environ.get('PORT', 8080))) + import os + + app.run(host="0.0.0.0", port=int(os.environ.get("PORT", 8080))) diff --git a/notifications_runtime.py b/notifications_runtime.py new file mode 100644 index 00000000..ebc4785f --- /dev/null +++ b/notifications_runtime.py @@ -0,0 +1,198 @@ +"""Runtime error / notification helpers for HTTP routes.""" + +from __future__ import annotations + + +import sys + +_FALLBACK_MODULE_NAME = __name__ + + +class _MainModuleProxy: + """Prefer ``main.`` after re-exports; fall back during module import.""" + + def __getattr__(self, name): + main_mod = sys.modules.get("main") + if main_mod is not None and name in getattr(main_mod, "__dict__", {}): + return main_mod.__dict__[name] + return getattr(sys.modules[_FALLBACK_MODULE_NAME], name) + + +main = _MainModuleProxy() + +import dataclasses +import json +import importlib +import os +import time +from datetime import datetime, timezone + +from flask import Flask, request + +import google.auth +import requests +from application.monitor_dispatcher import dispatch_due_monitors, load_monitor_targets +from application.startup_config import ( + StartupConfigReceipt, + normalize_monitor_dispatch_targets, + normalize_plugin_mounts_strategy, + overlay_env, +) +from application.broker_reconciliation import ( + account_snapshot_enabled, + collect_read_only_reconciliation_observations, + reconciliation_enabled, + run_read_only_account_snapshot, + run_read_only_broker_reconciliation, +) +from application.execution_receipt_adapter import ( + attach_cycle_execution_receipt, + attach_terminal_fallback_execution_receipt, +) +from application.runtime_broker_adapters import build_runtime_broker_adapters +from application.runtime_composer import build_runtime_composer +from application.rebalance_service import run_strategy as run_rebalance_cycle +from application.durable_execution_commands import ( + resolve_paper_execution_command_consumer_enabled, +) +from application.paper_execution_command_consumer import ( + consume_due_paper_execution_commands, +) +from application.runtime_strategy_adapters import build_runtime_strategy_adapters +from application.longbridge_execution import submit_order +from application import validation_cycle_observation +from runtime_execution_policy import fractional_buy_execution_enabled, FRACTIONAL_BUY_QUANTITY_STEP +from application.longbridge_portfolio import fetch_strategy_account_state +from application.account_snapshot import ( + begin_natural_cycle_history, cycle_history_observation, end_natural_cycle_history, + read_cycle_account_balance, read_cycle_stock_positions, +) +from scripts.record_daily_account_snapshot import record_projected_daily_account +from entrypoints.cloud_run import is_market_open_now +from runtime_execution_policy import dca_execution_unsupported_reason +from runtime_config_support import load_platform_runtime_settings +from strategy_registry import V7_LONGTERM_COMPOUNDING_PROFILE +from notifications.telegram import build_signal_text, build_strategy_display_name, build_translator +from notifications import unified as unified_notifications +from quant_platform_kit.common.runtime_reports import ( + append_runtime_report_error, + build_runtime_report_base, + finalize_runtime_report, + persist_runtime_report, +) +from quant_platform_kit.common.strategy_plugins import ( + build_strategy_plugin_report_payload, + load_configured_strategy_plugin_signals, + parse_strategy_plugin_mounts, +) +from quant_platform_kit.notifications.strategy_plugin_alerts import ( + StrategyPluginAlertStateSettings, + build_strategy_plugin_alert_context_label as build_alert_context_label, + publish_strategy_plugin_alerts as dispatch_strategy_plugin_alerts, +) +from quant_platform_kit.common.runtime_inputs import build_strategy_evaluation_inputs +from runtime_logging import build_run_id, emit_runtime_log +from quant_platform_kit.longbridge import ( + build_contexts, + calculate_rotation_indicators, + estimate_max_purchase_quantity, + fetch_last_price, + fetch_order_status, + fetch_token_from_secret, + refresh_token_if_needed, +) +from strategy_runtime import load_strategy_runtime +from decision_mapper import map_strategy_decision_to_plan + +def _runtime_failure_reason(exc: Exception) -> str: + if isinstance(exc, RuntimeError) and str(exc) == 'validation_journal_read_failed': + return 'validation_journal_read_failed' + if isinstance(exc, RuntimeError) and str(exc) == 'validation_journal_unavailable': + return 'validation_journal_unavailable' + if isinstance(exc, RuntimeError) and str(exc) == 'validation_journal_empty': + return 'validation_journal_empty' + if isinstance(exc, RuntimeError) and str(exc) == 'validation_side_effect_enabled': + return 'validation_side_effect_enabled' + if isinstance(exc, RuntimeError) and str(exc) == 'validation_marker_write_blocked': + return 'validation_marker_write_blocked' + if isinstance(exc, RuntimeError) and str(exc) == 'validation_read_only_bootstrap_unavailable': + return 'validation_read_only_bootstrap_unavailable' + if isinstance(exc, RuntimeError) and str(exc) in {'durable execution command producer is paper-only and cannot be enabled live', 'durable execution command consumer is paper-only and cannot be enabled live', 'strategy risk state persistence is paper-only and cannot be enabled live'}: + return 'paper_live_conflict' + return 'unknown' + +def _compact_error_notification(exc: Exception, *, title: str | None=None, prefix: str='', phase: str='strategy_cycle') -> str: + error_text = type(exc).__name__ + known_phases = {'market_hours', 'runtime_configuration', 'strategy_cycle', 'reporting', 'health_probe'} + safe_phase = phase if phase in known_phases else 'strategy_cycle' + reason = main._runtime_failure_reason(exc) + message = '\n'.join((title or main.t('error_title'), main.t('failure_phase', phase=main.t(f'failure_phase_{safe_phase}')), main.t(f'failure_reason_{reason}'), f'{prefix}{error_text}')) + if len(message) <= main.COMPACT_ERROR_NOTIFICATION_MAX_CHARS: + return message + return message[:main.COMPACT_ERROR_NOTIFICATION_MAX_CHARS - 1].rstrip() + '…' + +def _split_env_list(value: str | None) -> tuple[str, ...]: + return tuple((item.strip() for item in str(value or '').replace(';', ',').split(',') if item.strip())) + +def _runtime_error_notification_targets() -> tuple[tuple[str, str], ...]: + targets: list[tuple[str, str]] = [] + if main.TG_TOKEN and main.TG_CHAT_ID: + targets.append((main.TG_TOKEN, main.TG_CHAT_ID)) + seen: set[tuple[str, str]] = set() + unique_targets: list[tuple[str, str]] = [] + for target in targets: + if target in seen: + continue + seen.add(target) + unique_targets.append(target) + return tuple(unique_targets) + +def _runtime_error_notification_message(exc: Exception, *, route_label: str) -> str: + title = 'runtime_failure_title' + return '\n'.join((main.t(title), main.t('strategy_label', name=main.strategy_display_name or main.STRATEGY_PROFILE), main.t('runtime_failure_context', context=main.ACCOUNT_REGION), main.t('runtime_failure_result'), main.t('runtime_failure_action'))) + +def _send_runtime_error(exc: Exception, *, route_label: str) -> bool: + targets = main._runtime_error_notification_targets() + if not targets: + print(main.t('runtime_notification_missing_target'), flush=True) + return False + message = main._runtime_error_notification_message(exc, route_label=route_label) + outcomes = [] + for token, chat_id in targets: + try: + response = main.requests.post(f'https://api.telegram.org/bot{token}/sendMessage', json={'chat_id': chat_id, 'text': message}, timeout=10) + status_code = int(getattr(response, 'status_code', 200) or 200) + acknowledged = 200 <= status_code < 300 + load_payload = getattr(response, 'json', None) + payload = load_payload() if acknowledged and callable(load_payload) else None + if isinstance(payload, dict) and payload.get('ok') is False: + acknowledged = False + outcomes.append(acknowledged) + except Exception: + print(main.t('runtime_notification_delivery_failed'), flush=True) + outcomes.append(False) + return bool(outcomes) and all(outcomes) + +def _notify_runtime_error(exc: Exception, *, route_label: str) -> bool: + return main.unified_notifications.dispatch_immediate(lambda: main._send_runtime_error(exc, route_label=route_label), kind='real_error', incident=f'runtime_route:{route_label}:{main._runtime_failure_reason(exc)}', occurred_at=main.datetime.now(main.timezone.utc), run_id=main.build_run_id()) if main.unified_notifications.enabled() else main._send_runtime_error(exc, route_label=route_label) + +def _handle_route_runtime_error(exc: Exception, *, route_label: str): + print(main.t('runtime_failure_log'), flush=True) + if main.validation_cycle_observation.current() is None: + main._notify_runtime_error(exc, route_label=route_label) + return ('Error', 500) + +def _route_with_runtime_error_fallback(handler, *, success_body: str, route_label: str, **kwargs): + try: + if main.unified_notifications.enabled(): + with main.unified_notifications.bind_route(route_label): + result = handler(**kwargs) + else: + result = handler(**kwargs) + except Exception as exc: + return main._handle_route_runtime_error(exc, route_label=route_label) + if result is False: + return ('Error', 500) + if isinstance(result, tuple): + return result + return (success_body, 200) diff --git a/runtime_wiring.py b/runtime_wiring.py new file mode 100644 index 00000000..cd8db220 --- /dev/null +++ b/runtime_wiring.py @@ -0,0 +1,268 @@ +"""Runtime configuration, adapters, and composer wiring.""" + +from __future__ import annotations + + +import sys + +_FALLBACK_MODULE_NAME = __name__ + + +class _MainModuleProxy: + """Prefer ``main.`` after re-exports; fall back during module import.""" + + def __getattr__(self, name): + main_mod = sys.modules.get("main") + if main_mod is not None and name in getattr(main_mod, "__dict__", {}): + return main_mod.__dict__[name] + return getattr(sys.modules[_FALLBACK_MODULE_NAME], name) + + +main = _MainModuleProxy() + +import dataclasses +import json +import importlib +import os +import time +from datetime import datetime, timezone + +from flask import Flask, request + +import google.auth +import requests +from application.monitor_dispatcher import dispatch_due_monitors, load_monitor_targets +from application.startup_config import ( + StartupConfigReceipt, + normalize_monitor_dispatch_targets, + normalize_plugin_mounts_strategy, + overlay_env, +) +from application.broker_reconciliation import ( + account_snapshot_enabled, + collect_read_only_reconciliation_observations, + reconciliation_enabled, + run_read_only_account_snapshot, + run_read_only_broker_reconciliation, +) +from application.execution_receipt_adapter import ( + attach_cycle_execution_receipt, + attach_terminal_fallback_execution_receipt, +) +from application.runtime_broker_adapters import build_runtime_broker_adapters +from application.runtime_composer import build_runtime_composer +from application.rebalance_service import run_strategy as run_rebalance_cycle +from application.durable_execution_commands import ( + resolve_paper_execution_command_consumer_enabled, +) +from application.paper_execution_command_consumer import ( + consume_due_paper_execution_commands, +) +from application.runtime_strategy_adapters import build_runtime_strategy_adapters +from application.longbridge_execution import submit_order +from application import validation_cycle_observation +from runtime_execution_policy import fractional_buy_execution_enabled, FRACTIONAL_BUY_QUANTITY_STEP +from application.longbridge_portfolio import fetch_strategy_account_state +from application.account_snapshot import ( + begin_natural_cycle_history, cycle_history_observation, end_natural_cycle_history, + read_cycle_account_balance, read_cycle_stock_positions, +) +from scripts.record_daily_account_snapshot import record_projected_daily_account +from entrypoints.cloud_run import is_market_open_now +from runtime_execution_policy import dca_execution_unsupported_reason +from runtime_config_support import load_platform_runtime_settings +from strategy_registry import V7_LONGTERM_COMPOUNDING_PROFILE +from notifications.telegram import build_signal_text, build_strategy_display_name, build_translator +from notifications import unified as unified_notifications +from quant_platform_kit.common.runtime_reports import ( + append_runtime_report_error, + build_runtime_report_base, + finalize_runtime_report, + persist_runtime_report, +) +from quant_platform_kit.common.strategy_plugins import ( + build_strategy_plugin_report_payload, + load_configured_strategy_plugin_signals, + parse_strategy_plugin_mounts, +) +from quant_platform_kit.notifications.strategy_plugin_alerts import ( + StrategyPluginAlertStateSettings, + build_strategy_plugin_alert_context_label as build_alert_context_label, + publish_strategy_plugin_alerts as dispatch_strategy_plugin_alerts, +) +from quant_platform_kit.common.runtime_inputs import build_strategy_evaluation_inputs +from runtime_logging import build_run_id, emit_runtime_log +from quant_platform_kit.longbridge import ( + build_contexts, + calculate_rotation_indicators, + estimate_max_purchase_quantity, + fetch_last_price, + fetch_order_status, + fetch_token_from_secret, + refresh_token_if_needed, +) +from strategy_runtime import load_strategy_runtime +from decision_mapper import map_strategy_decision_to_plan + +def get_project_id(): + try: + _, project_id = main.google.auth.default() + return project_id + except Exception: + return main.os.getenv('GOOGLE_CLOUD_PROJECT') +RUNTIME_SETTINGS = load_platform_runtime_settings(project_id_resolver=get_project_id) +PROJECT_ID = RUNTIME_SETTINGS.project_id +SECRET_NAME = RUNTIME_SETTINGS.secret_name +ACCOUNT_PREFIX = RUNTIME_SETTINGS.account_prefix +STRATEGY_PROFILE = RUNTIME_SETTINGS.strategy_profile +STRATEGY_DISPLAY_NAME = RUNTIME_SETTINGS.strategy_display_name + +def _normalize_plugin_mounts_strategy(raw_mounts_json: str | None) -> str | None: + normalized, _corrections = main.normalize_plugin_mounts_strategy(raw_mounts_json, strategy_profile=main.STRATEGY_PROFILE) + return normalized + +def _adopt_startup_config(settings, env): + """Explicitly normalize startup config without mutating ``os.environ`` (B08). + + Returns ``(settings, monitor_dispatch_env_overrides, receipt)``. + """ + corrections = [] + if hasattr(settings, 'strategy_plugin_mounts_json'): + normalized, plugin_corrections = main.normalize_plugin_mounts_strategy(settings.strategy_plugin_mounts_json, strategy_profile=main.STRATEGY_PROFILE) + corrections.extend(plugin_corrections) + if normalized != settings.strategy_plugin_mounts_json: + if main.dataclasses.is_dataclass(settings) and (not isinstance(settings, type)): + settings = main.dataclasses.replace(settings, strategy_plugin_mounts_json=normalized) + else: + setattr(settings, 'strategy_plugin_mounts_json', normalized) + overrides, monitor_corrections = main.normalize_monitor_dispatch_targets(env, strategy_profile=main.STRATEGY_PROFILE) + corrections.extend(monitor_corrections) + receipt = main.StartupConfigReceipt(strategy_profile=main.STRATEGY_PROFILE, corrections=tuple(corrections), env_overrides=tuple(sorted(overrides))) + return (settings, overrides, receipt) +RUNTIME_SETTINGS, MONITOR_DISPATCH_ENV_OVERRIDES, STARTUP_CONFIG_RECEIPT = _adopt_startup_config(RUNTIME_SETTINGS, os.environ) + +def _load_monitor_dispatch_targets(): + if not main.MONITOR_DISPATCH_ENV_OVERRIDES: + return main.load_monitor_targets() + return main.load_monitor_targets(main.overlay_env(main.os.environ, main.MONITOR_DISPATCH_ENV_OVERRIDES)) +ACCOUNT_REGION = RUNTIME_SETTINGS.account_region +MARKET = RUNTIME_SETTINGS.market +MARKET_CALENDAR = RUNTIME_SETTINGS.market_calendar +MARKET_TIMEZONE = RUNTIME_SETTINGS.market_timezone +SYMBOL_SUFFIX = RUNTIME_SETTINGS.symbol_suffix +TRADING_CURRENCY = RUNTIME_SETTINGS.trading_currency +NOTIFY_LANG = RUNTIME_SETTINGS.notify_lang +TG_TOKEN = RUNTIME_SETTINGS.tg_token +TG_CHAT_ID = RUNTIME_SETTINGS.tg_chat_id +from quant_platform_kit.notifications.cycle_channel import resolve_cycle_channel_and_url +_NOTIFICATION_CHANNEL, _NOTIFICATION_WEBHOOK_URL = resolve_cycle_channel_and_url(explicit_channel=RUNTIME_SETTINGS.notification_channel, telegram_token=TG_TOKEN, telegram_chat_id=TG_CHAT_ID, wecom_url=RUNTIME_SETTINGS.wecom_webhook_url, dingtalk_url=RUNTIME_SETTINGS.dingtalk_webhook_url, feishu_url=RUNTIME_SETTINGS.feishu_webhook_url, serverchan_url=RUNTIME_SETTINGS.serverchan_webhook_url) +CASH_ONLY_EXECUTION = getattr(RUNTIME_SETTINGS, 'cash_only_execution', True) +STRATEGY_RUNTIME = load_strategy_runtime(STRATEGY_PROFILE, runtime_settings=RUNTIME_SETTINGS, logger=lambda message: print(message, flush=True)) +STRATEGY_RUNTIME_CONFIG = dict(STRATEGY_RUNTIME.merged_runtime_config) +MANAGED_SYMBOLS = STRATEGY_RUNTIME.managed_symbols +AVAILABLE_INPUTS = frozenset(STRATEGY_RUNTIME.runtime_adapter.available_inputs) +BENCHMARK_SYMBOL = str(STRATEGY_RUNTIME_CONFIG.get('benchmark_symbol', 'QQQ')) + +def _emit_v7_loaded_version_log() -> None: + """Emit one safe startup receipt for the disabled candidate revision.""" + if main.STRATEGY_PROFILE != main.V7_LONGTERM_COMPOUNDING_PROFILE: + return + from application.v7_paper_application import V7PaperApplicationError, build_v7_loaded_version_readback, load_v7_paper_application_binding + binding = load_v7_paper_application_binding(main.os.environ) + if binding is None: + return + try: + readback = build_v7_loaded_version_readback(binding, runtime_settings=main.RUNTIME_SETTINGS, strategy_runtime=main.STRATEGY_RUNTIME, runtime_env=main.os.environ) + print(main.json.dumps({'event': 'v7_paper_application_loaded', 'status': 'applied_paused', 'readback': readback}, ensure_ascii=True, separators=(',', ':')), flush=True) + except V7PaperApplicationError: + print(main.json.dumps({'event': 'v7_paper_application_loaded', 'status': 'uncertain'}, ensure_ascii=True, separators=(',', ':')), flush=True) +_emit_v7_loaded_version_log() +SIGNAL_EFFECTIVE_AFTER_TRADING_DAYS = getattr(getattr(STRATEGY_RUNTIME.runtime_adapter, 'runtime_policy', None), 'signal_effective_after_trading_days', None) +LIMIT_SELL_DISCOUNT = 0.995 +LIMIT_BUY_PREMIUM = 1.005 +DEFAULT_LIMIT_BUY_PREMIUM_BY_SYMBOL = {'SOXL': 1.015, 'TQQQ': 1.01} +ORDER_POLL_INTERVAL_SEC = 1 +ORDER_POLL_MAX_ATTEMPTS = 8 +TOKEN_REFRESH_THRESHOLD_DAYS = 30 +DEFAULT_SAFE_HAVEN_CASH_SUBSTITUTE_THRESHOLD_USD = 1000.0 +DEFAULT_MIN_ORDER_NOTIONAL_USD = 100.0 +_FRACTIONAL_BUY_EXECUTION = fractional_buy_execution_enabled(STRATEGY_PROFILE) + +def _profile_submit_order(t_ctx, symbol, **kwargs): + return main.submit_order(t_ctx, symbol, allow_fractional_shares=main._FRACTIONAL_BUY_EXECUTION, quantity_step=main.FRACTIONAL_BUY_QUANTITY_STEP if main._FRACTIONAL_BUY_EXECUTION else 1.0, **kwargs) + +def _profile_estimate_max_purchase_quantity(t_ctx, symbol, **kwargs): + return main.estimate_max_purchase_quantity(t_ctx, symbol, fractional_shares=main._FRACTIONAL_BUY_EXECUTION, **kwargs) +SEPARATOR = '━━━━━━━━━━━━━━━━━━' +COMPACT_ERROR_NOTIFICATION_MAX_CHARS = 3500 + +def _load_limit_buy_premium_by_symbol(*env_names: str) -> dict[str, float]: + raw_value = '' + for env_name in env_names: + value = main.os.getenv(env_name) + if value and value.strip(): + raw_value = value.strip() + break + if not raw_value: + return dict(main.DEFAULT_LIMIT_BUY_PREMIUM_BY_SYMBOL) + try: + payload = main.json.loads(raw_value) + except main.json.JSONDecodeError as exc: + raise ValueError(f'Invalid limit buy premium map JSON: {raw_value!r}') from exc + if not isinstance(payload, dict): + raise ValueError('Limit buy premium map must be a JSON object keyed by symbol.') + parsed: dict[str, float] = {} + for symbol, premium in payload.items(): + symbol_text = str(symbol or '').strip().upper() + if not symbol_text: + continue + premium_value = float(premium) + if premium_value <= 0.0: + raise ValueError(f'Limit buy premium for {symbol_text} must be positive.') + parsed[symbol_text] = premium_value + return parsed +LIMIT_BUY_PREMIUM_BY_SYMBOL = _load_limit_buy_premium_by_symbol('LONGBRIDGE_LIMIT_BUY_PREMIUM_BY_SYMBOL_JSON', 'LIMIT_BUY_PREMIUM_BY_SYMBOL_JSON') + +def t(key, **kwargs): + return main.build_translator(main.NOTIFY_LANG)(key, **kwargs) + +def _build_notification_delivery_log_for_report(*, platform: str, strategy_profile: str, run_id: str, dry_run: bool, orders_previewed_count: int, delivery_events: list[dict]) -> dict: + events = [dict(event) for event in delivery_events if dict(event).get('delivery_status') == 'sent'] + if not dry_run or orders_previewed_count <= 0 or (not events): + return {} + return {'notification_schema_version': 'hk_live_enablement_notification.v1', 'notification_event_type': 'hk_snapshot_live_enablement_dry_run', 'notification_correlation_id': str(run_id or ''), 'locales': ['en', 'zh-Hans'], 'profile': str(strategy_profile or ''), 'platform': str(platform or ''), 'validation_status': 'passed', 'orders_previewed': int(orders_previewed_count), 'delivery_events': events, 'notification_contains_profile': True, 'notification_contains_platform': True, 'notification_contains_validation_status': True, 'notification_contains_order_preview_summary': True, 'notification_redacts_sensitive_fields': True, 'redaction_policy': 'raw notification text is not persisted; only sha256 and length are recorded'} + +def _build_notification_delivery_summary(delivery_events: list[dict]) -> dict: + safe_fields = ('sink', 'delivery_status', 'transport_acknowledged', 'error_type', 'compact_text_sha256', 'compact_text_length') + events = [{key: event[key] for key in safe_fields if key in event} for event in (dict(item) for item in delivery_events)] + if not events: + return {} + sent_count = sum((event.get('delivery_status') == 'sent' for event in events)) + failed_count = sum((event.get('delivery_status') == 'failed' for event in events)) + return {'attempted_count': len(events), 'sent_count': sent_count, 'failed_count': failed_count, 'all_acknowledged': failed_count == 0 and sent_count == len(events), 'delivery_events': events} +signal_text = build_signal_text(t) +strategy_display_name = build_strategy_display_name(t)(STRATEGY_PROFILE, fallback_name=STRATEGY_DISPLAY_NAME, metadata=RUNTIME_SETTINGS.strategy_metadata, locale=NOTIFY_LANG) + +def log_position_snapshot(message): + print(f'[{main.ACCOUNT_REGION}] {message}', flush=True) + +def log_runtime_warning(message): + print(f'[{main.ACCOUNT_REGION}] [warning] {message}', flush=True) +BROKER_ADAPTERS = build_runtime_broker_adapters(strategy_symbols=tuple(MANAGED_SYMBOLS), account_hash=ACCOUNT_PREFIX or ACCOUNT_REGION or 'longbridge', fetch_last_price_fn=fetch_last_price, fetch_strategy_account_state_fn=lambda quote_context, trade_context: fetch_strategy_account_state(quote_context, trade_context, list(MANAGED_SYMBOLS), cash_currency=TRADING_CURRENCY, position_log_fn=log_position_snapshot if getattr(RUNTIME_SETTINGS, 'debug_position_snapshot', False) else None, warning_log_fn=log_runtime_warning), submit_order_fn=_profile_submit_order, symbol_suffix=SYMBOL_SUFFIX, currency=TRADING_CURRENCY, cash_only_execution=CASH_ONLY_EXECUTION) +STRATEGY_ADAPTERS = build_runtime_strategy_adapters(strategy_runtime=STRATEGY_RUNTIME, strategy_profile=STRATEGY_PROFILE, strategy_runtime_config=STRATEGY_RUNTIME_CONFIG, available_inputs=AVAILABLE_INPUTS, benchmark_symbol=BENCHMARK_SYMBOL, signal_text_fn=signal_text, translator=t, broker_adapters=BROKER_ADAPTERS, calculate_rotation_indicators_fn=calculate_rotation_indicators, build_strategy_evaluation_inputs_fn=build_strategy_evaluation_inputs, map_strategy_decision_to_plan_fn=map_strategy_decision_to_plan, execution_policy={'reserved_cash_floor_usd': getattr(RUNTIME_SETTINGS, 'reserved_cash_floor_usd', 0.0), 'reserved_cash_ratio': getattr(RUNTIME_SETTINGS, 'reserved_cash_ratio', 0.0), 'cash_only_execution': CASH_ONLY_EXECUTION}, execution_materials=getattr(STRATEGY_RUNTIME, 'execution_materials', {}), build_strategy_plugin_report_payload_fn=build_strategy_plugin_report_payload, load_configured_strategy_plugin_signals_fn=load_configured_strategy_plugin_signals, parse_strategy_plugin_mounts_fn=parse_strategy_plugin_mounts) + +def _safe_haven_cash_substitute_threshold_usd() -> float: + return float(getattr(main.RUNTIME_SETTINGS, 'safe_haven_cash_substitute_threshold_usd', main.DEFAULT_SAFE_HAVEN_CASH_SUBSTITUTE_THRESHOLD_USD)) + +def build_composer(*, dry_run_only_override: bool | None=None): + return main.build_runtime_composer(project_id=main.PROJECT_ID, secret_name=main.SECRET_NAME, token_refresh_threshold_days=main.TOKEN_REFRESH_THRESHOLD_DAYS, account_prefix=main.ACCOUNT_PREFIX, account_region=main.ACCOUNT_REGION, strategy_profile=main.STRATEGY_PROFILE, strategy_display_name=main.STRATEGY_DISPLAY_NAME, strategy_display_name_localized=main.strategy_display_name, strategy_domain=main.RUNTIME_SETTINGS.strategy_domain, notify_lang=main.NOTIFY_LANG, tg_token=main.TG_TOKEN, tg_chat_id=main.TG_CHAT_ID, notification_channel=main._NOTIFICATION_CHANNEL, webhook_url=main._NOTIFICATION_WEBHOOK_URL, managed_symbols=tuple(main.MANAGED_SYMBOLS), benchmark_symbol=main.BENCHMARK_SYMBOL, signal_effective_after_trading_days=main.SIGNAL_EFFECTIVE_AFTER_TRADING_DAYS, separator=main.SEPARATOR, limit_sell_discount=main.LIMIT_SELL_DISCOUNT, limit_buy_premium=main.LIMIT_BUY_PREMIUM, limit_buy_premium_by_symbol=main.LIMIT_BUY_PREMIUM_BY_SYMBOL, order_poll_interval_sec=main.ORDER_POLL_INTERVAL_SEC, order_poll_max_attempts=main.ORDER_POLL_MAX_ATTEMPTS, safe_haven_cash_substitute_threshold_usd=main._safe_haven_cash_substitute_threshold_usd(), min_order_notional_usd=main._min_order_notional_usd(), market=main.MARKET, symbol_suffix=main.SYMBOL_SUFFIX, trading_currency=main.TRADING_CURRENCY, dry_run_only=main.RUNTIME_SETTINGS.dry_run_only, dry_run_only_override=dry_run_only_override, broker_adapters=main.BROKER_ADAPTERS, strategy_adapters=main.STRATEGY_ADAPTERS, estimate_max_purchase_quantity_fn=main._profile_estimate_max_purchase_quantity, fetch_order_status_fn=main.fetch_order_status, fetch_token_from_secret_fn=main.fetch_token_from_secret, fetch_token_with_metadata_fn=main.fetch_account_snapshot_token_with_metadata, refresh_token_if_needed_fn=main.refresh_token_if_needed, build_contexts_fn=main.build_contexts, run_id_builder=main.build_run_id, event_logger=main.emit_runtime_log, report_builder=main.build_runtime_report_base, report_persister=main.persist_runtime_report, translator=main.t, runtime_target=main.RUNTIME_SETTINGS.runtime_target, env_reader=main.os.getenv, sleeper=main.time.sleep, printer=print, extra_reporting_fields={'market': main.MARKET, 'market_calendar': main.MARKET_CALENDAR, 'market_timezone': main.MARKET_TIMEZONE, 'symbol_suffix': main.SYMBOL_SUFFIX, 'trading_currency': main.TRADING_CURRENCY}) + +def fetch_account_snapshot_token_with_metadata(project_id, secret_name): + """Read one configured secret-store response. A metadata failure is not reread.""" + from types import SimpleNamespace + from quant_platform_kit.cloud import get_secret_store + store = get_secret_store() + read_metadata = getattr(store, 'get_secret_with_metadata', None) + if callable(read_metadata): + return read_metadata(secret_name, project_id=project_id) + return SimpleNamespace(value=store.get_secret(secret_name, project_id=project_id), version_name=None) diff --git a/strategy_plugins_runtime.py b/strategy_plugins_runtime.py new file mode 100644 index 00000000..e4a146fc --- /dev/null +++ b/strategy_plugins_runtime.py @@ -0,0 +1,124 @@ +"""Strategy plugin signal presentation and alert dispatch.""" + +from __future__ import annotations + + +import sys + +_FALLBACK_MODULE_NAME = __name__ + + +class _MainModuleProxy: + """Prefer ``main.`` after re-exports; fall back during module import.""" + + def __getattr__(self, name): + main_mod = sys.modules.get("main") + if main_mod is not None and name in getattr(main_mod, "__dict__", {}): + return main_mod.__dict__[name] + return getattr(sys.modules[_FALLBACK_MODULE_NAME], name) + + +main = _MainModuleProxy() + +import dataclasses +import json +import importlib +import os +import time +from datetime import datetime, timezone + +from flask import Flask, request + +import google.auth +import requests +from application.monitor_dispatcher import dispatch_due_monitors, load_monitor_targets +from application.startup_config import ( + StartupConfigReceipt, + normalize_monitor_dispatch_targets, + normalize_plugin_mounts_strategy, + overlay_env, +) +from application.broker_reconciliation import ( + account_snapshot_enabled, + collect_read_only_reconciliation_observations, + reconciliation_enabled, + run_read_only_account_snapshot, + run_read_only_broker_reconciliation, +) +from application.execution_receipt_adapter import ( + attach_cycle_execution_receipt, + attach_terminal_fallback_execution_receipt, +) +from application.runtime_broker_adapters import build_runtime_broker_adapters +from application.runtime_composer import build_runtime_composer +from application.rebalance_service import run_strategy as run_rebalance_cycle +from application.durable_execution_commands import ( + resolve_paper_execution_command_consumer_enabled, +) +from application.paper_execution_command_consumer import ( + consume_due_paper_execution_commands, +) +from application.runtime_strategy_adapters import build_runtime_strategy_adapters +from application.longbridge_execution import submit_order +from application import validation_cycle_observation +from runtime_execution_policy import fractional_buy_execution_enabled, FRACTIONAL_BUY_QUANTITY_STEP +from application.longbridge_portfolio import fetch_strategy_account_state +from application.account_snapshot import ( + begin_natural_cycle_history, cycle_history_observation, end_natural_cycle_history, + read_cycle_account_balance, read_cycle_stock_positions, +) +from scripts.record_daily_account_snapshot import record_projected_daily_account +from entrypoints.cloud_run import is_market_open_now +from runtime_execution_policy import dca_execution_unsupported_reason +from runtime_config_support import load_platform_runtime_settings +from strategy_registry import V7_LONGTERM_COMPOUNDING_PROFILE +from notifications.telegram import build_signal_text, build_strategy_display_name, build_translator +from notifications import unified as unified_notifications +from quant_platform_kit.common.runtime_reports import ( + append_runtime_report_error, + build_runtime_report_base, + finalize_runtime_report, + persist_runtime_report, +) +from quant_platform_kit.common.strategy_plugins import ( + build_strategy_plugin_report_payload, + load_configured_strategy_plugin_signals, + parse_strategy_plugin_mounts, +) +from quant_platform_kit.notifications.strategy_plugin_alerts import ( + StrategyPluginAlertStateSettings, + build_strategy_plugin_alert_context_label as build_alert_context_label, + publish_strategy_plugin_alerts as dispatch_strategy_plugin_alerts, +) +from quant_platform_kit.common.runtime_inputs import build_strategy_evaluation_inputs +from runtime_logging import build_run_id, emit_runtime_log +from quant_platform_kit.longbridge import ( + build_contexts, + calculate_rotation_indicators, + estimate_max_purchase_quantity, + fetch_last_price, + fetch_order_status, + fetch_token_from_secret, + refresh_token_if_needed, +) +from strategy_runtime import load_strategy_runtime +from decision_mapper import map_strategy_decision_to_plan + +def build_strategy_plugin_alert_messages(signals): + return main.STRATEGY_ADAPTERS.build_strategy_plugin_alert_messages(signals) + +def build_strategy_plugin_alert_state_settings(): + return main.StrategyPluginAlertStateSettings.from_env(project_id=main.PROJECT_ID) + +def build_strategy_plugin_alert_context_label() -> str: + return main.build_alert_context_label(platform_id='longbridge', strategy_profile=main.STRATEGY_PROFILE, account_scope=main.ACCOUNT_REGION, service_name=main.SECRET_NAME, runtime_target=main.RUNTIME_SETTINGS.runtime_target) + +def publish_strategy_plugin_alerts(signals, *, report=None): + from notifications.digest_snapshot.digest_routing import immediate_plugin_signals + from notifications.digest_snapshot.digest_candidates import routine_observation_counts + if report is not None: + report.setdefault('summary', {})['routine_observation_counts'] = routine_observation_counts(signals) + result = main.dispatch_strategy_plugin_alerts(immediate_plugin_signals(signals, mode='daily_digest'), notification_settings=main.RUNTIME_SETTINGS, translator=main.t, strategy_label=main.STRATEGY_PROFILE, context_label=main.build_strategy_plugin_alert_context_label(), state_settings=main.build_strategy_plugin_alert_state_settings(), log_message=print) + if report is not None: + result.attach_to_report(report) + return result From cc8c0850c791847ad7c9646649c4633fcca30950 Mon Sep 17 00:00:00 2001 From: Pigbibi <20649888+Pigbibi@users.noreply.github.com> Date: Sat, 14 Nov 2026 04:33:41 +0800 Subject: [PATCH 2/7] fix: thin main facade and update notification source scans Align LongBridge entry with explicit re-export facade and point i18n extract tests at notifications_runtime after the split. Co-authored-by: Cursor --- http_routes.py | 10 +- main.py | 311 ++++++++---------------- notifications_runtime.py | 10 +- runtime_wiring.py | 10 +- strategy_plugins_runtime.py | 10 +- tests/test_runtime_notification_i18n.py | 12 +- 6 files changed, 139 insertions(+), 224 deletions(-) diff --git a/http_routes.py b/http_routes.py index b6b3b371..51e55dad 100644 --- a/http_routes.py +++ b/http_routes.py @@ -9,12 +9,16 @@ class _MainModuleProxy: - """Prefer ``main.`` after re-exports; fall back during module import.""" + """Prefer ``main.`` (facade-aware); fall back during module import.""" def __getattr__(self, name): main_mod = sys.modules.get("main") - if main_mod is not None and name in getattr(main_mod, "__dict__", {}): - return main_mod.__dict__[name] + if main_mod is not None: + # Use getattr so ModuleType facades can resolve mutable globals. + try: + return getattr(main_mod, name) + except AttributeError: + pass return getattr(sys.modules[_FALLBACK_MODULE_NAME], name) diff --git a/main.py b/main.py index 02093335..e8685df8 100644 --- a/main.py +++ b/main.py @@ -1,9 +1,17 @@ -"""Platform HTTP entrypoint with explicit named re-exports.""" +"""LongBridgePlatform HTTP entrypoint. +Thin facade over the platform split modules. Public helpers are +re-exported by explicit name so existing ``import main`` / monkeypatch +call sites keep working. Mutable process globals resolve through the +module facade so rebinding inside implementation modules stays visible. +""" from __future__ import annotations import sys +import types +# Reload split modules whenever main is (re)imported so test harnesses +# that only pop ``main`` still observe fresh env/stubs. for _split_module_name in ( "runtime_wiring", "notifications_runtime", @@ -13,120 +21,18 @@ sys.modules.pop(_split_module_name, None) del _split_module_name -import dataclasses -import json -import importlib -import os -import time -from datetime import datetime, timezone - -from flask import Flask, request - -import google.auth -import requests -from application.monitor_dispatcher import dispatch_due_monitors, load_monitor_targets -from application.startup_config import ( - StartupConfigReceipt, - normalize_monitor_dispatch_targets, - normalize_plugin_mounts_strategy, - overlay_env, -) -from application.broker_reconciliation import ( - account_snapshot_enabled, - collect_read_only_reconciliation_observations, - reconciliation_enabled, - run_read_only_account_snapshot, - run_read_only_broker_reconciliation, -) -from application.execution_receipt_adapter import ( - attach_cycle_execution_receipt, - attach_terminal_fallback_execution_receipt, -) -from application.runtime_broker_adapters import build_runtime_broker_adapters -from application.runtime_composer import build_runtime_composer -from application.rebalance_service import run_strategy as run_rebalance_cycle -from application.durable_execution_commands import ( - resolve_paper_execution_command_consumer_enabled, -) -from application.paper_execution_command_consumer import ( - consume_due_paper_execution_commands, -) -from application.runtime_strategy_adapters import build_runtime_strategy_adapters -from application.longbridge_execution import submit_order -from application import validation_cycle_observation -from runtime_execution_policy import fractional_buy_execution_enabled, FRACTIONAL_BUY_QUANTITY_STEP -from application.longbridge_portfolio import fetch_strategy_account_state -from application.account_snapshot import ( - begin_natural_cycle_history, cycle_history_observation, end_natural_cycle_history, - read_cycle_account_balance, read_cycle_stock_positions, -) -from scripts.record_daily_account_snapshot import record_projected_daily_account -from entrypoints.cloud_run import is_market_open_now -from runtime_execution_policy import dca_execution_unsupported_reason -from runtime_config_support import load_platform_runtime_settings -from strategy_registry import V7_LONGTERM_COMPOUNDING_PROFILE -from notifications.telegram import build_signal_text, build_strategy_display_name, build_translator -from notifications import unified as unified_notifications -from quant_platform_kit.common.runtime_reports import ( - append_runtime_report_error, - build_runtime_report_base, - finalize_runtime_report, - persist_runtime_report, -) -from quant_platform_kit.common.strategy_plugins import ( - build_strategy_plugin_report_payload, - load_configured_strategy_plugin_signals, - parse_strategy_plugin_mounts, -) -from quant_platform_kit.notifications.strategy_plugin_alerts import ( - StrategyPluginAlertStateSettings, - build_strategy_plugin_alert_context_label as build_alert_context_label, - publish_strategy_plugin_alerts as dispatch_strategy_plugin_alerts, -) -from quant_platform_kit.common.runtime_inputs import build_strategy_evaluation_inputs -from runtime_logging import build_run_id, emit_runtime_log -from quant_platform_kit.longbridge import ( - build_contexts, - calculate_rotation_indicators, - estimate_max_purchase_quantity, - fetch_last_price, - fetch_order_status, - fetch_token_from_secret, - refresh_token_if_needed, -) -from strategy_runtime import load_strategy_runtime -from decision_mapper import map_strategy_decision_to_plan +import http_routes as _http_routes +import notifications_runtime as _notifications_runtime +import runtime_wiring as _runtime_wiring +import strategy_plugins_runtime as _strategy_plugins_runtime from runtime_wiring import ( get_project_id, - RUNTIME_SETTINGS, - PROJECT_ID, - SECRET_NAME, - ACCOUNT_PREFIX, - STRATEGY_PROFILE, - STRATEGY_DISPLAY_NAME, _normalize_plugin_mounts_strategy, _adopt_startup_config, - MONITOR_DISPATCH_ENV_OVERRIDES, - STARTUP_CONFIG_RECEIPT, _load_monitor_dispatch_targets, - ACCOUNT_REGION, - MARKET, - MARKET_CALENDAR, - MARKET_TIMEZONE, - SYMBOL_SUFFIX, - TRADING_CURRENCY, - NOTIFY_LANG, - TG_TOKEN, - TG_CHAT_ID, _NOTIFICATION_CHANNEL, _NOTIFICATION_WEBHOOK_URL, - CASH_ONLY_EXECUTION, - STRATEGY_RUNTIME, - STRATEGY_RUNTIME_CONFIG, - MANAGED_SYMBOLS, - AVAILABLE_INPUTS, - BENCHMARK_SYMBOL, _emit_v7_loaded_version_log, SIGNAL_EFFECTIVE_AFTER_TRADING_DAYS, LIMIT_SELL_DISCOUNT, @@ -143,7 +49,6 @@ SEPARATOR, COMPACT_ERROR_NOTIFICATION_MAX_CHARS, _load_limit_buy_premium_by_symbol, - LIMIT_BUY_PREMIUM_BY_SYMBOL, t, _build_notification_delivery_log_for_report, _build_notification_delivery_summary, @@ -151,8 +56,6 @@ strategy_display_name, log_position_snapshot, log_runtime_warning, - BROKER_ADAPTERS, - STRATEGY_ADAPTERS, _safe_haven_cash_substitute_threshold_usd, build_composer, fetch_account_snapshot_token_with_metadata, @@ -179,7 +82,6 @@ from http_routes import ( app, - READ_ONLY_BROKER_RECONCILIATION_COLLECTOR, _summarize_cycle_result_for_report, _min_order_notional_usd, run_strategy, @@ -206,107 +108,12 @@ ) __all__ = [ - 'FRACTIONAL_BUY_QUANTITY_STEP', - 'Flask', - 'StartupConfigReceipt', - 'StrategyPluginAlertStateSettings', - 'V7_LONGTERM_COMPOUNDING_PROFILE', - 'account_snapshot_enabled', - 'append_runtime_report_error', - 'attach_cycle_execution_receipt', - 'attach_terminal_fallback_execution_receipt', - 'begin_natural_cycle_history', - 'build_alert_context_label', - 'build_contexts', - 'build_run_id', - 'build_runtime_broker_adapters', - 'build_runtime_composer', - 'build_runtime_report_base', - 'build_runtime_strategy_adapters', - 'build_signal_text', - 'build_strategy_display_name', - 'build_strategy_evaluation_inputs', - 'build_strategy_plugin_report_payload', - 'build_translator', - 'calculate_rotation_indicators', - 'collect_read_only_reconciliation_observations', - 'consume_due_paper_execution_commands', - 'cycle_history_observation', - 'dataclasses', - 'datetime', - 'dca_execution_unsupported_reason', - 'dispatch_due_monitors', - 'dispatch_strategy_plugin_alerts', - 'emit_runtime_log', - 'end_natural_cycle_history', - 'estimate_max_purchase_quantity', - 'fetch_last_price', - 'fetch_order_status', - 'fetch_strategy_account_state', - 'fetch_token_from_secret', - 'finalize_runtime_report', - 'fractional_buy_execution_enabled', - 'google', - 'importlib', - 'is_market_open_now', - 'json', - 'load_configured_strategy_plugin_signals', - 'load_monitor_targets', - 'load_platform_runtime_settings', - 'load_strategy_runtime', - 'map_strategy_decision_to_plan', - 'normalize_monitor_dispatch_targets', - 'normalize_plugin_mounts_strategy', - 'os', - 'overlay_env', - 'parse_strategy_plugin_mounts', - 'persist_runtime_report', - 'read_cycle_account_balance', - 'read_cycle_stock_positions', - 'reconciliation_enabled', - 'record_projected_daily_account', - 'refresh_token_if_needed', - 'request', - 'requests', - 'resolve_cycle_channel_and_url', - 'resolve_paper_execution_command_consumer_enabled', - 'run_read_only_account_snapshot', - 'run_read_only_broker_reconciliation', - 'run_rebalance_cycle', - 'submit_order', - 'time', - 'timezone', - 'unified_notifications', - 'validation_cycle_observation', 'get_project_id', - 'RUNTIME_SETTINGS', - 'PROJECT_ID', - 'SECRET_NAME', - 'ACCOUNT_PREFIX', - 'STRATEGY_PROFILE', - 'STRATEGY_DISPLAY_NAME', '_normalize_plugin_mounts_strategy', '_adopt_startup_config', - 'MONITOR_DISPATCH_ENV_OVERRIDES', - 'STARTUP_CONFIG_RECEIPT', '_load_monitor_dispatch_targets', - 'ACCOUNT_REGION', - 'MARKET', - 'MARKET_CALENDAR', - 'MARKET_TIMEZONE', - 'SYMBOL_SUFFIX', - 'TRADING_CURRENCY', - 'NOTIFY_LANG', - 'TG_TOKEN', - 'TG_CHAT_ID', '_NOTIFICATION_CHANNEL', '_NOTIFICATION_WEBHOOK_URL', - 'CASH_ONLY_EXECUTION', - 'STRATEGY_RUNTIME', - 'STRATEGY_RUNTIME_CONFIG', - 'MANAGED_SYMBOLS', - 'AVAILABLE_INPUTS', - 'BENCHMARK_SYMBOL', '_emit_v7_loaded_version_log', 'SIGNAL_EFFECTIVE_AFTER_TRADING_DAYS', 'LIMIT_SELL_DISCOUNT', @@ -323,7 +130,6 @@ 'SEPARATOR', 'COMPACT_ERROR_NOTIFICATION_MAX_CHARS', '_load_limit_buy_premium_by_symbol', - 'LIMIT_BUY_PREMIUM_BY_SYMBOL', 't', '_build_notification_delivery_log_for_report', '_build_notification_delivery_summary', @@ -331,11 +137,35 @@ 'strategy_display_name', 'log_position_snapshot', 'log_runtime_warning', - 'BROKER_ADAPTERS', - 'STRATEGY_ADAPTERS', '_safe_haven_cash_substitute_threshold_usd', 'build_composer', 'fetch_account_snapshot_token_with_metadata', + 'RUNTIME_SETTINGS', + 'PROJECT_ID', + 'SECRET_NAME', + 'ACCOUNT_PREFIX', + 'STRATEGY_PROFILE', + 'STRATEGY_DISPLAY_NAME', + 'MONITOR_DISPATCH_ENV_OVERRIDES', + 'STARTUP_CONFIG_RECEIPT', + 'ACCOUNT_REGION', + 'MARKET', + 'MARKET_CALENDAR', + 'MARKET_TIMEZONE', + 'SYMBOL_SUFFIX', + 'TRADING_CURRENCY', + 'NOTIFY_LANG', + 'TG_TOKEN', + 'TG_CHAT_ID', + 'CASH_ONLY_EXECUTION', + 'STRATEGY_RUNTIME', + 'STRATEGY_RUNTIME_CONFIG', + 'MANAGED_SYMBOLS', + 'AVAILABLE_INPUTS', + 'BENCHMARK_SYMBOL', + 'LIMIT_BUY_PREMIUM_BY_SYMBOL', + 'BROKER_ADAPTERS', + 'STRATEGY_ADAPTERS', '_runtime_failure_reason', '_compact_error_notification', '_split_env_list', @@ -350,7 +180,6 @@ 'build_strategy_plugin_alert_context_label', 'publish_strategy_plugin_alerts', 'app', - 'READ_ONLY_BROKER_RECONCILIATION_COLLECTOR', '_summarize_cycle_result_for_report', '_min_order_notional_usd', 'run_strategy', @@ -374,8 +203,70 @@ 'handle_monitor_dispatch', 'request_method', 'health', + 'READ_ONLY_BROKER_RECONCILIATION_COLLECTOR', ] +_IMPL_MODULES = ( + _runtime_wiring, + _notifications_runtime, + _strategy_plugins_runtime, + _http_routes, +) + +# Keep these off this module's __dict__ so reads see live impl bindings. +_MUTABLE_PROCESS_GLOBALS = frozenset({ + 'ACCOUNT_PREFIX', + 'ACCOUNT_REGION', + 'AVAILABLE_INPUTS', + 'BENCHMARK_SYMBOL', + 'BROKER_ADAPTERS', + 'CASH_ONLY_EXECUTION', + 'LIMIT_BUY_PREMIUM_BY_SYMBOL', + 'MANAGED_SYMBOLS', + 'MARKET', + 'MARKET_CALENDAR', + 'MARKET_TIMEZONE', + 'MONITOR_DISPATCH_ENV_OVERRIDES', + 'NOTIFY_LANG', + 'PROJECT_ID', + 'READ_ONLY_BROKER_RECONCILIATION_COLLECTOR', + 'RUNTIME_SETTINGS', + 'SECRET_NAME', + 'STARTUP_CONFIG_RECEIPT', + 'STRATEGY_ADAPTERS', + 'STRATEGY_DISPLAY_NAME', + 'STRATEGY_PROFILE', + 'STRATEGY_RUNTIME', + 'STRATEGY_RUNTIME_CONFIG', + 'SYMBOL_SUFFIX', + 'TG_CHAT_ID', + 'TG_TOKEN', + 'TRADING_CURRENCY', +}) + +class _LongBridgeMainFacade(types.ModuleType): + """Forward attribute writes to the owning implementation module.""" + + def __getattr__(self, name: str): + for module in _IMPL_MODULES: + if name in module.__dict__: + return module.__dict__[name] + raise AttributeError(f"module {__name__!r} has no attribute {name!r}") + + def __setattr__(self, name: str, value): + if name in _MUTABLE_PROCESS_GLOBALS: + for module in _IMPL_MODULES: + if name in module.__dict__: + setattr(module, name, value) + return + raise AttributeError(f"module {__name__!r} has no attribute {name!r}") + super().__setattr__(name, value) + for module in _IMPL_MODULES: + if name in module.__dict__: + setattr(module, name, value) + +sys.modules[__name__].__class__ = _LongBridgeMainFacade + if __name__ == "__main__": import os diff --git a/notifications_runtime.py b/notifications_runtime.py index ebc4785f..addc68f6 100644 --- a/notifications_runtime.py +++ b/notifications_runtime.py @@ -9,12 +9,16 @@ class _MainModuleProxy: - """Prefer ``main.`` after re-exports; fall back during module import.""" + """Prefer ``main.`` (facade-aware); fall back during module import.""" def __getattr__(self, name): main_mod = sys.modules.get("main") - if main_mod is not None and name in getattr(main_mod, "__dict__", {}): - return main_mod.__dict__[name] + if main_mod is not None: + # Use getattr so ModuleType facades can resolve mutable globals. + try: + return getattr(main_mod, name) + except AttributeError: + pass return getattr(sys.modules[_FALLBACK_MODULE_NAME], name) diff --git a/runtime_wiring.py b/runtime_wiring.py index cd8db220..cf23f52b 100644 --- a/runtime_wiring.py +++ b/runtime_wiring.py @@ -9,12 +9,16 @@ class _MainModuleProxy: - """Prefer ``main.`` after re-exports; fall back during module import.""" + """Prefer ``main.`` (facade-aware); fall back during module import.""" def __getattr__(self, name): main_mod = sys.modules.get("main") - if main_mod is not None and name in getattr(main_mod, "__dict__", {}): - return main_mod.__dict__[name] + if main_mod is not None: + # Use getattr so ModuleType facades can resolve mutable globals. + try: + return getattr(main_mod, name) + except AttributeError: + pass return getattr(sys.modules[_FALLBACK_MODULE_NAME], name) diff --git a/strategy_plugins_runtime.py b/strategy_plugins_runtime.py index e4a146fc..37088cb6 100644 --- a/strategy_plugins_runtime.py +++ b/strategy_plugins_runtime.py @@ -9,12 +9,16 @@ class _MainModuleProxy: - """Prefer ``main.`` after re-exports; fall back during module import.""" + """Prefer ``main.`` (facade-aware); fall back during module import.""" def __getattr__(self, name): main_mod = sys.modules.get("main") - if main_mod is not None and name in getattr(main_mod, "__dict__", {}): - return main_mod.__dict__[name] + if main_mod is not None: + # Use getattr so ModuleType facades can resolve mutable globals. + try: + return getattr(main_mod, name) + except AttributeError: + pass return getattr(sys.modules[_FALLBACK_MODULE_NAME], name) diff --git a/tests/test_runtime_notification_i18n.py b/tests/test_runtime_notification_i18n.py index 577eb4df..e79b9624 100644 --- a/tests/test_runtime_notification_i18n.py +++ b/tests/test_runtime_notification_i18n.py @@ -14,7 +14,15 @@ def test_chinese_locale_variants_use_chinese(locale): def runtime_function(name, **overrides): - tree = ast.parse((Path(__file__).resolve().parents[1] / "main.py").read_text()) + root = Path(__file__).resolve().parents[1] + source_name = { + "_runtime_error_notification_message": "notifications_runtime.py", + "_runtime_error_notification_targets": "notifications_runtime.py", + "_notify_runtime_error": "notifications_runtime.py", + "_publish_runtime_failure_notification": "notifications_runtime.py", + "_publish_runtime_failure_notification_legacy": "notifications_runtime.py", + }.get(name, "main.py") + tree = ast.parse((root / source_name).read_text()) function = next(node for node in tree.body if isinstance(node, ast.FunctionDef) and node.name == name) namespace = dict(os=os, request=SimpleNamespace(method="GET", path="/main"), NOTIFY_LANG="zh-CN", SERVICE_NAME="Test service", SECRET_NAME="Test secret alias", @@ -22,7 +30,7 @@ def runtime_function(name, **overrides): strategy_display_name="Test strategy", build_translator=build_translator, t=build_translator("zh-CN")) namespace.update(overrides) - exec(compile(ast.Module(body=[function], type_ignores=[]), "main.py", "exec"), namespace) + exec(compile(ast.Module(body=[function], type_ignores=[]), source_name, "exec"), namespace) return namespace[name] From 3a59303a7cac57814fe1f5f67d626768e022e6d3 Mon Sep 17 00:00:00 2001 From: Pigbibi <20649888+Pigbibi@users.noreply.github.com> Date: Sat, 14 Nov 2026 04:34:25 +0800 Subject: [PATCH 3/7] refactor: thin main facade with explicit re-exports Align with Binance Wave2: main is a named-export module facade only, with Flask/wiring living in split modules. Co-authored-by: Cursor --- main.py | 124 +++++++++++++++++++++++++------------------------------- 1 file changed, 56 insertions(+), 68 deletions(-) diff --git a/main.py b/main.py index e8685df8..30b4ab85 100644 --- a/main.py +++ b/main.py @@ -2,16 +2,14 @@ Thin facade over the platform split modules. Public helpers are re-exported by explicit name so existing ``import main`` / monkeypatch -call sites keep working. Mutable process globals resolve through the -module facade so rebinding inside implementation modules stays visible. +call sites keep working. A module facade forwards attribute writes to +the owning implementation module. """ from __future__ import annotations import sys import types -# Reload split modules whenever main is (re)imported so test harnesses -# that only pop ``main`` still observe fresh env/stubs. for _split_module_name in ( "runtime_wiring", "notifications_runtime", @@ -28,11 +26,34 @@ from runtime_wiring import ( get_project_id, + RUNTIME_SETTINGS, + PROJECT_ID, + SECRET_NAME, + ACCOUNT_PREFIX, + STRATEGY_PROFILE, + STRATEGY_DISPLAY_NAME, _normalize_plugin_mounts_strategy, _adopt_startup_config, + MONITOR_DISPATCH_ENV_OVERRIDES, + STARTUP_CONFIG_RECEIPT, _load_monitor_dispatch_targets, + ACCOUNT_REGION, + MARKET, + MARKET_CALENDAR, + MARKET_TIMEZONE, + SYMBOL_SUFFIX, + TRADING_CURRENCY, + NOTIFY_LANG, + TG_TOKEN, + TG_CHAT_ID, _NOTIFICATION_CHANNEL, _NOTIFICATION_WEBHOOK_URL, + CASH_ONLY_EXECUTION, + STRATEGY_RUNTIME, + STRATEGY_RUNTIME_CONFIG, + MANAGED_SYMBOLS, + AVAILABLE_INPUTS, + BENCHMARK_SYMBOL, _emit_v7_loaded_version_log, SIGNAL_EFFECTIVE_AFTER_TRADING_DAYS, LIMIT_SELL_DISCOUNT, @@ -49,6 +70,7 @@ SEPARATOR, COMPACT_ERROR_NOTIFICATION_MAX_CHARS, _load_limit_buy_premium_by_symbol, + LIMIT_BUY_PREMIUM_BY_SYMBOL, t, _build_notification_delivery_log_for_report, _build_notification_delivery_summary, @@ -56,6 +78,8 @@ strategy_display_name, log_position_snapshot, log_runtime_warning, + BROKER_ADAPTERS, + STRATEGY_ADAPTERS, _safe_haven_cash_substitute_threshold_usd, build_composer, fetch_account_snapshot_token_with_metadata, @@ -82,6 +106,7 @@ from http_routes import ( app, + READ_ONLY_BROKER_RECONCILIATION_COLLECTOR, _summarize_cycle_result_for_report, _min_order_notional_usd, run_strategy, @@ -109,11 +134,34 @@ __all__ = [ 'get_project_id', + 'RUNTIME_SETTINGS', + 'PROJECT_ID', + 'SECRET_NAME', + 'ACCOUNT_PREFIX', + 'STRATEGY_PROFILE', + 'STRATEGY_DISPLAY_NAME', '_normalize_plugin_mounts_strategy', '_adopt_startup_config', + 'MONITOR_DISPATCH_ENV_OVERRIDES', + 'STARTUP_CONFIG_RECEIPT', '_load_monitor_dispatch_targets', + 'ACCOUNT_REGION', + 'MARKET', + 'MARKET_CALENDAR', + 'MARKET_TIMEZONE', + 'SYMBOL_SUFFIX', + 'TRADING_CURRENCY', + 'NOTIFY_LANG', + 'TG_TOKEN', + 'TG_CHAT_ID', '_NOTIFICATION_CHANNEL', '_NOTIFICATION_WEBHOOK_URL', + 'CASH_ONLY_EXECUTION', + 'STRATEGY_RUNTIME', + 'STRATEGY_RUNTIME_CONFIG', + 'MANAGED_SYMBOLS', + 'AVAILABLE_INPUTS', + 'BENCHMARK_SYMBOL', '_emit_v7_loaded_version_log', 'SIGNAL_EFFECTIVE_AFTER_TRADING_DAYS', 'LIMIT_SELL_DISCOUNT', @@ -130,6 +178,7 @@ 'SEPARATOR', 'COMPACT_ERROR_NOTIFICATION_MAX_CHARS', '_load_limit_buy_premium_by_symbol', + 'LIMIT_BUY_PREMIUM_BY_SYMBOL', 't', '_build_notification_delivery_log_for_report', '_build_notification_delivery_summary', @@ -137,35 +186,11 @@ 'strategy_display_name', 'log_position_snapshot', 'log_runtime_warning', + 'BROKER_ADAPTERS', + 'STRATEGY_ADAPTERS', '_safe_haven_cash_substitute_threshold_usd', 'build_composer', 'fetch_account_snapshot_token_with_metadata', - 'RUNTIME_SETTINGS', - 'PROJECT_ID', - 'SECRET_NAME', - 'ACCOUNT_PREFIX', - 'STRATEGY_PROFILE', - 'STRATEGY_DISPLAY_NAME', - 'MONITOR_DISPATCH_ENV_OVERRIDES', - 'STARTUP_CONFIG_RECEIPT', - 'ACCOUNT_REGION', - 'MARKET', - 'MARKET_CALENDAR', - 'MARKET_TIMEZONE', - 'SYMBOL_SUFFIX', - 'TRADING_CURRENCY', - 'NOTIFY_LANG', - 'TG_TOKEN', - 'TG_CHAT_ID', - 'CASH_ONLY_EXECUTION', - 'STRATEGY_RUNTIME', - 'STRATEGY_RUNTIME_CONFIG', - 'MANAGED_SYMBOLS', - 'AVAILABLE_INPUTS', - 'BENCHMARK_SYMBOL', - 'LIMIT_BUY_PREMIUM_BY_SYMBOL', - 'BROKER_ADAPTERS', - 'STRATEGY_ADAPTERS', '_runtime_failure_reason', '_compact_error_notification', '_split_env_list', @@ -180,6 +205,7 @@ 'build_strategy_plugin_alert_context_label', 'publish_strategy_plugin_alerts', 'app', + 'READ_ONLY_BROKER_RECONCILIATION_COLLECTOR', '_summarize_cycle_result_for_report', '_min_order_notional_usd', 'run_strategy', @@ -203,7 +229,6 @@ 'handle_monitor_dispatch', 'request_method', 'health', - 'READ_ONLY_BROKER_RECONCILIATION_COLLECTOR', ] _IMPL_MODULES = ( @@ -213,37 +238,6 @@ _http_routes, ) -# Keep these off this module's __dict__ so reads see live impl bindings. -_MUTABLE_PROCESS_GLOBALS = frozenset({ - 'ACCOUNT_PREFIX', - 'ACCOUNT_REGION', - 'AVAILABLE_INPUTS', - 'BENCHMARK_SYMBOL', - 'BROKER_ADAPTERS', - 'CASH_ONLY_EXECUTION', - 'LIMIT_BUY_PREMIUM_BY_SYMBOL', - 'MANAGED_SYMBOLS', - 'MARKET', - 'MARKET_CALENDAR', - 'MARKET_TIMEZONE', - 'MONITOR_DISPATCH_ENV_OVERRIDES', - 'NOTIFY_LANG', - 'PROJECT_ID', - 'READ_ONLY_BROKER_RECONCILIATION_COLLECTOR', - 'RUNTIME_SETTINGS', - 'SECRET_NAME', - 'STARTUP_CONFIG_RECEIPT', - 'STRATEGY_ADAPTERS', - 'STRATEGY_DISPLAY_NAME', - 'STRATEGY_PROFILE', - 'STRATEGY_RUNTIME', - 'STRATEGY_RUNTIME_CONFIG', - 'SYMBOL_SUFFIX', - 'TG_CHAT_ID', - 'TG_TOKEN', - 'TRADING_CURRENCY', -}) - class _LongBridgeMainFacade(types.ModuleType): """Forward attribute writes to the owning implementation module.""" @@ -254,12 +248,6 @@ def __getattr__(self, name: str): raise AttributeError(f"module {__name__!r} has no attribute {name!r}") def __setattr__(self, name: str, value): - if name in _MUTABLE_PROCESS_GLOBALS: - for module in _IMPL_MODULES: - if name in module.__dict__: - setattr(module, name, value) - return - raise AttributeError(f"module {__name__!r} has no attribute {name!r}") super().__setattr__(name, value) for module in _IMPL_MODULES: if name in module.__dict__: From 76962f514f79b91a5b0ff46b64387ea1e32d2740 Mon Sep 17 00:00:00 2001 From: Pigbibi <20649888+Pigbibi@users.noreply.github.com> Date: Sat, 14 Nov 2026 04:39:42 +0800 Subject: [PATCH 4/7] fix: resolve __main__ entry and include split modules in wheel http_routes entry proxy now accepts script launch as __main__, and py-modules lists the extracted runtime modules for installed imports. Co-authored-by: Cursor --- http_routes.py | 17 +++++++++++++---- pyproject.toml | 4 ++++ 2 files changed, 17 insertions(+), 4 deletions(-) diff --git a/http_routes.py b/http_routes.py index 51e55dad..f04a80eb 100644 --- a/http_routes.py +++ b/http_routes.py @@ -12,13 +12,22 @@ class _MainModuleProxy: """Prefer ``main.`` (facade-aware); fall back during module import.""" def __getattr__(self, name): - main_mod = sys.modules.get("main") - if main_mod is not None: - # Use getattr so ModuleType facades can resolve mutable globals. + for key in ("main", "__main__"): + main_mod = sys.modules.get(key) + if main_mod is None: + continue + if getattr(main_mod, "__name__", None) in { + "http_routes", + "runtime_wiring", + "notifications_runtime", + "strategy_plugins_runtime", + }: + continue try: + # Use getattr so ModuleType facades can resolve mutable globals. return getattr(main_mod, name) except AttributeError: - pass + continue return getattr(sys.modules[_FALLBACK_MODULE_NAME], name) diff --git a/pyproject.toml b/pyproject.toml index e09b7eb1..47423faf 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -41,11 +41,15 @@ show_missing = true [tool.setuptools] py-modules = [ "decision_mapper", + "http_routes", "main", + "notifications_runtime", "runtime_config_support", "runtime_execution_policy", "runtime_logging", + "runtime_wiring", "strategy_loader", + "strategy_plugins_runtime", "strategy_registry", "strategy_runtime", ] From 874c513e2c85c65f8fa23176d11fe7f5de3f2aec Mon Sep 17 00:00:00 2001 From: Pigbibi <20649888+Pigbibi@users.noreply.github.com> Date: Sat, 14 Nov 2026 04:40:17 +0800 Subject: [PATCH 5/7] fix: accept __main__ in notification proxy and AST harness Script launches and facade-qualified notification helpers need the same entrypoint resolution and a main stand-in for source extraction tests. Co-authored-by: Cursor --- notifications_runtime.py | 17 +++++++++++++---- tests/test_runtime_notification_i18n.py | 9 +++++++++ 2 files changed, 22 insertions(+), 4 deletions(-) diff --git a/notifications_runtime.py b/notifications_runtime.py index addc68f6..122dddcf 100644 --- a/notifications_runtime.py +++ b/notifications_runtime.py @@ -12,13 +12,22 @@ class _MainModuleProxy: """Prefer ``main.`` (facade-aware); fall back during module import.""" def __getattr__(self, name): - main_mod = sys.modules.get("main") - if main_mod is not None: - # Use getattr so ModuleType facades can resolve mutable globals. + for key in ("main", "__main__"): + main_mod = sys.modules.get(key) + if main_mod is None: + continue + if getattr(main_mod, "__name__", None) in { + "http_routes", + "runtime_wiring", + "notifications_runtime", + "strategy_plugins_runtime", + }: + continue try: + # Use getattr so ModuleType facades can resolve mutable globals. return getattr(main_mod, name) except AttributeError: - pass + continue return getattr(sys.modules[_FALLBACK_MODULE_NAME], name) diff --git a/tests/test_runtime_notification_i18n.py b/tests/test_runtime_notification_i18n.py index e79b9624..69371a1b 100644 --- a/tests/test_runtime_notification_i18n.py +++ b/tests/test_runtime_notification_i18n.py @@ -30,6 +30,15 @@ def runtime_function(name, **overrides): strategy_display_name="Test strategy", build_translator=build_translator, t=build_translator("zh-CN")) namespace.update(overrides) + + class _MainProxy: + def __getattr__(self, attr): + try: + return namespace[attr] + except KeyError as exc: + raise AttributeError(attr) from exc + + namespace["main"] = _MainProxy() exec(compile(ast.Module(body=[function], type_ignores=[]), source_name, "exec"), namespace) return namespace[name] From 8ca4d7a3437ee97272a04699bb38b6cf2a82a39e Mon Sep 17 00:00:00 2001 From: Pigbibi <20649888+Pigbibi@users.noreply.github.com> Date: Sat, 14 Nov 2026 04:43:02 +0800 Subject: [PATCH 6/7] fix: load archive finalizer summarize helper from http_routes Co-authored-by: Cursor --- tests/archive_finalizer_support.py | 7 +++++-- 1 file changed, 5 insertions(+), 2 deletions(-) diff --git a/tests/archive_finalizer_support.py b/tests/archive_finalizer_support.py index e2fdaf08..24716c23 100644 --- a/tests/archive_finalizer_support.py +++ b/tests/archive_finalizer_support.py @@ -217,10 +217,13 @@ def sender(**kw): with unified.bind_report(report): returned = run(runtime=runtime, config=config) assert returned is result + class _MainProxy: + unified_notifications = unified + summarize = function_from( - (ROOT / "main.py").read_text(), + (ROOT / "http_routes.py").read_text(), "_summarize_cycle_result_for_report", - {"unified_notifications": unified}, + {"unified_notifications": unified, "main": _MainProxy()}, ) summary = summarize(result, dry_run=False) if report is not None: From 239a5bebece0b872856f8ab0c05d10215d4eb62e Mon Sep 17 00:00:00 2001 From: Pigbibi <20649888+Pigbibi@users.noreply.github.com> Date: Sat, 14 Nov 2026 04:48:47 +0800 Subject: [PATCH 7/7] fix: keep facade patches local so originals restore Stop forwarding setattr into implementation modules so patch.object deletion returns __getattr__ to the real callables. Co-authored-by: Cursor --- main.py | 17 +++++++++++------ 1 file changed, 11 insertions(+), 6 deletions(-) diff --git a/main.py b/main.py index 30b4ab85..35b15ad1 100644 --- a/main.py +++ b/main.py @@ -239,7 +239,12 @@ ) class _LongBridgeMainFacade(types.ModuleType): - """Forward attribute writes to the owning implementation module.""" + """Resolve missing attributes from split modules without mutating them. + + ``unittest.mock.patch.object`` restores by deleting temporary facade + attributes. Keeping writes local to the facade lets ``__getattr__`` return + the original implementation binding after the patch context exits. + """ def __getattr__(self, name: str): for module in _IMPL_MODULES: @@ -247,11 +252,11 @@ def __getattr__(self, name: str): return module.__dict__[name] raise AttributeError(f"module {__name__!r} has no attribute {name!r}") - def __setattr__(self, name: str, value): - super().__setattr__(name, value) - for module in _IMPL_MODULES: - if name in module.__dict__: - setattr(module, name, value) + def __delattr__(self, name: str): + if name in self.__dict__: + super().__delattr__(name) + return + raise AttributeError(f"module {__name__!r} has no attribute {name!r}") sys.modules[__name__].__class__ = _LongBridgeMainFacade