From 1955f45880246d38f7b7736b0e806d509b67406c Mon Sep 17 00:00:00 2001 From: Pigbibi <20649888+Pigbibi@users.noreply.github.com> Date: Tue, 21 Apr 2026 18:19:27 +0800 Subject: [PATCH] decouple ibkr runtime wiring --- application/cycle_result.py | 25 + application/rebalance_service.py | 222 ++++---- application/runtime_broker_adapters.py | 275 ++++++++++ application/runtime_composer.py | 220 ++++++++ application/runtime_dependencies.py | 27 + application/runtime_notification_adapters.py | 37 ++ application/runtime_reporting_adapters.py | 227 ++++++++ application/runtime_strategy_adapters.py | 81 +++ main.py | 474 +++++++---------- notifications/renderers.py | 518 +++++++++++++++++++ tests/test_rebalance_service.py | 10 +- tests/test_request_handling.py | 17 +- tests/test_runtime_composer.py | 90 ++++ 13 files changed, 1810 insertions(+), 413 deletions(-) create mode 100644 application/cycle_result.py create mode 100644 application/runtime_broker_adapters.py create mode 100644 application/runtime_composer.py create mode 100644 application/runtime_dependencies.py create mode 100644 application/runtime_notification_adapters.py create mode 100644 application/runtime_reporting_adapters.py create mode 100644 application/runtime_strategy_adapters.py create mode 100644 notifications/renderers.py create mode 100644 tests/test_runtime_composer.py diff --git a/application/cycle_result.py b/application/cycle_result.py new file mode 100644 index 0000000..0a8990b --- /dev/null +++ b/application/cycle_result.py @@ -0,0 +1,25 @@ +"""Structured strategy cycle results for InteractiveBrokersPlatform.""" + +from __future__ import annotations + +from dataclasses import dataclass, field +from typing import Any + + +@dataclass(frozen=True) +class StrategyCycleResult: + """Structured output from one strategy cycle.""" + + result: str + signal_metadata: dict[str, Any] = field(default_factory=dict) + target_weights: dict[str, float] | None = None + execution_summary: dict[str, Any] = field(default_factory=dict) + reconciliation_record: dict[str, Any] = field(default_factory=dict) + reconciliation_record_path: str | None = None + + +def coerce_strategy_cycle_result(value: StrategyCycleResult | str) -> StrategyCycleResult: + """Keep request-handling tolerant of older string-only test doubles.""" + if isinstance(value, StrategyCycleResult): + return value + return StrategyCycleResult(result=str(value)) diff --git a/application/rebalance_service.py b/application/rebalance_service.py index 3d72b91..d1cc09e 100644 --- a/application/rebalance_service.py +++ b/application/rebalance_service.py @@ -3,14 +3,20 @@ from __future__ import annotations from collections.abc import Mapping +from datetime import datetime, timezone import json import re +from application.cycle_result import StrategyCycleResult +from application.runtime_dependencies import IBKRRebalanceConfig, IBKRRebalanceRuntime from application.reconciliation_service import ( build_reconciliation_record, write_reconciliation_record, ) -from notifications.events import NotificationPublisher, RenderedNotification +from notifications.events import NotificationPublisher +from notifications import renderers as notification_renderers +from quant_platform_kit.common.models import PortfolioSnapshot, Position +from quant_platform_kit.common.port_adapters import CallableNotificationPort, CallablePortfolioPort _ZH_REASON_REPLACEMENTS = ( @@ -443,29 +449,86 @@ def _build_compact_message( return "\n".join(lines) +def _legacy_portfolio_snapshot(ib, *, get_current_portfolio) -> PortfolioSnapshot: + positions, account_values = get_current_portfolio(ib) + snapshot_positions = tuple( + Position( + symbol=str(symbol).strip().upper(), + quantity=float(details.get("quantity") or 0), + market_value=float(details.get("quantity") or 0) * float(details.get("avg_cost") or 0.0), + average_cost=float(details.get("avg_cost") or 0.0), + ) + for symbol, details in dict(positions or {}).items() + ) + return PortfolioSnapshot( + as_of=datetime.now(timezone.utc), + total_equity=float(account_values.get("equity") or 0.0), + buying_power=float(account_values.get("buying_power") or 0.0), + positions=snapshot_positions, + ) + + +def _snapshot_to_portfolio_view(snapshot) -> tuple[dict[str, dict[str, float | int]], dict[str, float]]: + positions = {} + for position in getattr(snapshot, "positions", ()) or (): + positions[str(position.symbol).strip().upper()] = { + "quantity": int(position.quantity), + "avg_cost": float(position.average_cost or 0.0), + } + account_values = { + "equity": float(getattr(snapshot, "total_equity", 0.0) or 0.0), + "buying_power": float(getattr(snapshot, "buying_power", 0.0) or 0.0), + } + return positions, account_values + + def run_strategy_core( *, - connect_ib, - get_current_portfolio, - compute_signals, - execute_rebalance, - send_tg_message, - translator, - separator, + runtime: IBKRRebalanceRuntime | None = None, + config: IBKRRebalanceConfig | None = None, + connect_ib=None, + get_current_portfolio=None, + compute_signals=None, + execute_rebalance=None, + send_tg_message=None, + translator=None, + separator=None, strategy_display_name=None, reconciliation_output_path=None, - result_hook=None, ): + if runtime is None: + if not all((connect_ib, get_current_portfolio, compute_signals, execute_rebalance, send_tg_message)): + raise ValueError("Legacy IBKR rebalance call requires connect_ib/get_current_portfolio/compute_signals/execute_rebalance/send_tg_message") + runtime = IBKRRebalanceRuntime( + connect_ib=connect_ib, + portfolio_port_factory=lambda ib: CallablePortfolioPort( + lambda: _legacy_portfolio_snapshot(ib, get_current_portfolio=get_current_portfolio) + ), + compute_signals=compute_signals, + execute_rebalance=execute_rebalance, + notifications=CallableNotificationPort(send_tg_message), + ) + if config is None: + if translator is None or separator is None: + raise ValueError("IBKR rebalance config requires translator and separator") + config = IBKRRebalanceConfig( + translator=translator, + separator=separator, + strategy_display_name=strategy_display_name, + reconciliation_output_path=reconciliation_output_path, + ) + notification_publisher = NotificationPublisher( log_message=lambda message: print(message, flush=True), - send_message=send_tg_message, + send_message=runtime.notifications.send_text, ) ib = None try: - ib = connect_ib() - positions, account_values = get_current_portfolio(ib) + ib = runtime.connect_ib() + snapshot = runtime.portfolio_port_factory(ib).get_portfolio_snapshot() + positions, account_values = _snapshot_to_portfolio_view(snapshot) current_holdings = set(positions.keys()) - signal_result = compute_signals(ib, current_holdings) + signal_result = runtime.compute_signals(ib, current_holdings) if len(signal_result) == 5: target_weights, signal_desc, _is_emergency, status_desc, signal_metadata = signal_result else: @@ -474,17 +537,17 @@ def run_strategy_core( allocation = _resolve_weight_allocation(signal_metadata, required=target_weights is not None) resolved_target_weights = dict(allocation.get("targets") or {}) if target_weights is not None else None - dashboard = build_dashboard( + dashboard = notification_renderers.build_dashboard( positions, account_values, signal_desc, status_desc, strategy_profile=signal_metadata.get("strategy_profile"), - strategy_display_name=strategy_display_name, + strategy_display_name=config.strategy_display_name, target_weights=resolved_target_weights, signal_metadata=signal_metadata, - translator=translator, - separator=separator, + translator=config.translator, + separator=config.separator, status_icon=signal_metadata.get("status_icon", "🐤"), ) strategy_dashboard = _strategy_dashboard_text(signal_metadata) @@ -493,13 +556,13 @@ def run_strategy_core( decision = signal_metadata.get("snapshot_guard_decision") no_op_reason = signal_metadata.get("no_op_reason") fail_reason = signal_metadata.get("fail_reason") - no_op_text = translator("no_trades") + no_op_text = config.translator("no_trades") if decision: - no_op_text = f"{no_op_text} | {_localize_notification_text(f'decision={decision}', translator=translator)}" + no_op_text = f"{no_op_text} | {_localize_notification_text(f'decision={decision}', translator=config.translator)}" if no_op_reason: - no_op_text = f"{no_op_text} | {_localize_notification_text(f'reason={no_op_reason}', translator=translator)}" + no_op_text = f"{no_op_text} | {_localize_notification_text(f'reason={no_op_reason}', translator=config.translator)}" if fail_reason: - no_op_text = f"{no_op_text} | {_localize_notification_text(f'fail_reason={fail_reason}', translator=translator)}" + no_op_text = f"{no_op_text} | {_localize_notification_text(f'fail_reason={fail_reason}', translator=config.translator)}" no_op_text = "\n".join(_split_labeled_text(no_op_text)) record = build_reconciliation_record( strategy_profile=signal_metadata.get("strategy_profile"), @@ -511,44 +574,35 @@ def run_strategy_core( execution_summary=None, no_op_reason=no_op_reason or fail_reason or decision, ) - record_path = write_reconciliation_record(record, output_path=reconciliation_output_path) + record_path = write_reconciliation_record(record, output_path=config.reconciliation_output_path) print( "reconciliation_record " + json.dumps({"path": str(record_path), "status": record.get("execution_status"), "no_op_reason": record.get("no_op_reason")}, ensure_ascii=False), flush=True, ) - detailed_message = f"{translator('heartbeat_title')}\n{dashboard}\n{separator}\n{no_op_text}" - compact_message = _build_compact_message( - title=translator("heartbeat_title"), - strategy_display_name=strategy_display_name, - signal_desc=signal_desc, - status_desc=status_desc, - status_icon=signal_metadata.get("status_icon", "🐤"), - translator=translator, - separator=separator, - body_lines=[no_op_text], - dashboard_text=strategy_dashboard, - ) notification_publisher.publish( - RenderedNotification( - detailed_text=detailed_message, - compact_text=compact_message, + notification_renderers.render_heartbeat_notification( + dashboard=dashboard, + strategy_dashboard=strategy_dashboard, + no_op_text=no_op_text, + signal_desc=signal_desc, + status_desc=status_desc, + status_icon=signal_metadata.get("status_icon", "🐤"), + translator=config.translator, + separator=config.separator, + strategy_display_name=config.strategy_display_name, ) ) - if callable(result_hook): - result_hook( - { - "result": "OK - heartbeat", - "signal_metadata": dict(signal_metadata or {}), - "target_weights": None, - "execution_summary": None, - "reconciliation_record": dict(record), - "reconciliation_record_path": str(record_path), - } - ) - return "OK - heartbeat" + return StrategyCycleResult( + result="OK - heartbeat", + signal_metadata=dict(signal_metadata or {}), + target_weights=None, + execution_summary={}, + reconciliation_record=dict(record), + reconciliation_record_path=str(record_path), + ) - execution_result = execute_rebalance( + execution_result = runtime.execute_rebalance( ib, resolved_target_weights, positions, @@ -570,7 +624,7 @@ def run_strategy_core( target_weights=resolved_target_weights, execution_summary=execution_summary, ) - record_path = write_reconciliation_record(record, output_path=reconciliation_output_path) + record_path = write_reconciliation_record(record, output_path=config.reconciliation_output_path) print( "reconciliation_record " + json.dumps( @@ -585,62 +639,28 @@ def run_strategy_core( ), flush=True, ) - if trade_logs: - notification_trade_lines = _build_notification_trade_lines( - trade_logs, + notification_publisher.publish( + notification_renderers.render_trade_notification( + dashboard=dashboard, + strategy_dashboard=strategy_dashboard, + trade_logs=trade_logs, execution_summary=execution_summary, - translator=translator, - ) - trade_lines = "\n".join(notification_trade_lines) - detailed_message = ( - f"{translator('rebalance_title')}\n" - f"{dashboard}\n" - f"{separator}\n" - f"{trade_lines}" - ) - compact_message = _build_compact_message( - title=translator("rebalance_title"), - strategy_display_name=strategy_display_name, - signal_desc=signal_desc, - status_desc=status_desc, - status_icon=signal_metadata.get("status_icon", "🐤"), - translator=translator, - separator=separator, - body_lines=notification_trade_lines, - dashboard_text=strategy_dashboard, - ) - else: - detailed_message = f"{translator('heartbeat_title')}\n{dashboard}\n{separator}\n{translator('no_trades')}" - compact_message = _build_compact_message( - title=translator("heartbeat_title"), - strategy_display_name=strategy_display_name, signal_desc=signal_desc, status_desc=status_desc, status_icon=signal_metadata.get("status_icon", "🐤"), - translator=translator, - separator=separator, - body_lines=[translator("no_trades")], - dashboard_text=strategy_dashboard, - ) - - notification_publisher.publish( - RenderedNotification( - detailed_text=detailed_message, - compact_text=compact_message, + translator=config.translator, + separator=config.separator, + strategy_display_name=config.strategy_display_name, ) ) - if callable(result_hook): - result_hook( - { - "result": "OK - executed", - "signal_metadata": dict(signal_metadata or {}), - "target_weights": dict(resolved_target_weights or {}), - "execution_summary": dict(execution_summary or {}), - "reconciliation_record": dict(record), - "reconciliation_record_path": str(record_path), - } - ) - return "OK - executed" + return StrategyCycleResult( + result="OK - executed", + signal_metadata=dict(signal_metadata or {}), + target_weights=dict(resolved_target_weights or {}), + execution_summary=dict(execution_summary or {}), + reconciliation_record=dict(record), + reconciliation_record_path=str(record_path), + ) finally: if ib is not None and ib.isConnected(): ib.disconnect() diff --git a/application/runtime_broker_adapters.py b/application/runtime_broker_adapters.py new file mode 100644 index 0000000..623b9bc --- /dev/null +++ b/application/runtime_broker_adapters.py @@ -0,0 +1,275 @@ +"""Builder helpers for IBKR broker-side runtime adapters.""" + +from __future__ import annotations + +from dataclasses import dataclass +from datetime import datetime, timezone +from typing import Any + +from application.cycle_result import StrategyCycleResult +from quant_platform_kit.common.models import OrderIntent, PortfolioSnapshot, Position + + +@dataclass(frozen=True) +class IBKRRuntimeBrokerAdapters: + host_resolver: Any + ib_port: int + ib_client_id: int + connect_timeout_seconds: int + connect_attempts: int + connect_retry_delay_seconds: float + client_id_retry_offset: int + ensure_event_loop_fn: Any + connect_ib_fn: Any + fetch_portfolio_snapshot_fn: Any + fetch_quote_snapshots_fn: Any + submit_order_intent_fn: Any + order_intent_cls: Any + application_get_market_prices_fn: Any + application_check_order_submitted_fn: Any + application_execute_rebalance_fn: Any + execute_paper_liquidation_fn: Any + translator: Any + strategy_profile: str + account_group: str + service_name: str | None + account_ids: tuple[str, ...] + dry_run_only: bool + cash_reserve_ratio: float + rebalance_threshold_ratio: float + limit_buy_premium: float + sell_settle_delay_sec: float + separator: str + strategy_display_name: str + sleep_fn: Any + printer: Any = print + + def connect_ib(self): + self.ensure_event_loop_fn() + host = self.host_resolver() + last_error = None + for attempt in range(1, self.connect_attempts + 1): + client_id = self.ib_client_id + ((attempt - 1) * self.client_id_retry_offset) + self.printer( + "Connecting to IB gateway " + f"{host}:{self.ib_port} " + f"(client_id={client_id}, " + f"attempt={attempt}/{self.connect_attempts}, " + f"timeout={self.connect_timeout_seconds}s)", + flush=True, + ) + try: + return self.connect_ib_fn( + host, + self.ib_port, + client_id, + timeout=self.connect_timeout_seconds, + ) + except (ConnectionError, TimeoutError, OSError) as exc: + last_error = exc + self.printer( + "IB gateway connection attempt failed " + f"(attempt={attempt}/{self.connect_attempts}, " + f"client_id={client_id}, " + f"error_type={type(exc).__name__}, " + f"error={exc})", + flush=True, + ) + if attempt < self.connect_attempts and self.connect_retry_delay_seconds > 0: + self.sleep_fn(self.connect_retry_delay_seconds) + raise last_error + + def get_current_portfolio(self, ib): + snapshot = self.fetch_portfolio_snapshot_fn(ib) + positions = {} + for position in snapshot.positions: + positions[position.symbol] = { + "quantity": int(position.quantity), + "avg_cost": float(position.average_cost or 0.0), + } + account_values = { + "equity": snapshot.total_equity, + "buying_power": snapshot.buying_power or 0.0, + } + return positions, account_values + + def build_portfolio_snapshot(self, ib, *, get_current_portfolio_fallback=None): + if hasattr(ib, "reqPositions"): + return self.fetch_portfolio_snapshot_fn(ib) + positions, account_values = get_current_portfolio_fallback(ib) + return PortfolioSnapshot( + as_of=datetime.now(timezone.utc), + total_equity=float(account_values.get("equity") or 0.0), + buying_power=float(account_values.get("buying_power") or 0.0), + positions=tuple( + Position( + symbol=str(symbol).strip().upper(), + quantity=float(details.get("quantity") or 0), + market_value=float(details.get("quantity") or 0) * float(details.get("avg_cost") or 0.0), + average_cost=float(details.get("avg_cost") or 0.0), + ) + for symbol, details in dict(positions or {}).items() + ), + ) + + def get_market_prices(self, ib, symbols): + return self.application_get_market_prices_fn( + ib, + symbols, + fetch_quote_snapshots=self.fetch_quote_snapshots_fn, + ) + + def check_order_submitted(self, report): + return self.application_check_order_submitted_fn(report, translator=self.translator) + + def execute_rebalance( + self, + ib, + target_weights, + positions, + account_values, + *, + strategy_symbols=None, + signal_metadata=None, + ): + return self.application_execute_rebalance_fn( + ib, + target_weights, + positions, + account_values, + fetch_quote_snapshots=self.fetch_quote_snapshots_fn, + submit_order_intent=self.submit_order_intent_fn, + order_intent_cls=self.order_intent_cls, + translator=self.translator, + strategy_symbols=strategy_symbols, + signal_metadata=signal_metadata or {}, + strategy_profile=self.strategy_profile, + account_group=self.account_group, + service_name=self.service_name, + account_ids=self.account_ids, + dry_run_only=self.dry_run_only, + cash_reserve_ratio=self.cash_reserve_ratio, + rebalance_threshold_ratio=self.rebalance_threshold_ratio, + limit_buy_premium=self.limit_buy_premium, + sell_settle_delay_sec=self.sell_settle_delay_sec, + return_summary=True, + ) + + def format_liquidation_orders(self, orders) -> str: + preview = [] + for order in orders or (): + symbol = str(order.get("symbol") or "").strip().upper() + side = str(order.get("side") or "").strip().lower() + quantity = float(order.get("quantity") or 0.0) + status = str(order.get("status") or "").strip() + if symbol: + preview.append(f"{symbol} {side} {quantity:g} {status}".strip()) + return ", ".join(preview) if preview else self.translator("no_trades") + + def run_paper_liquidation_cycle( + self, + *, + connect_ib_fn, + get_current_portfolio_fn, + publish_notification_fn, + ): + ib = connect_ib_fn() + try: + positions, _account_values = get_current_portfolio_fn(ib) + if not positions: + self.printer("paper_liquidation_positions_empty_retry", flush=True) + self.sleep_fn(2.0) + positions, _account_values = get_current_portfolio_fn(ib) + summary = self.execute_paper_liquidation_fn( + ib, + positions, + submit_order_intent=self.submit_order_intent_fn, + order_intent_cls=self.order_intent_cls, + dry_run_only=self.dry_run_only, + ) + message = ( + f"{self.translator('rebalance_title')}\n" + f"{self.translator('strategy_label', name=self.strategy_display_name)}\n" + f"{self.translator('paper_liquidation_only')}\n" + f"{self.translator('paper_liquidation_status', mode=summary['mode'], status=summary['execution_status'])}\n" + f"{self.translator('paper_liquidation_positions_seen', count=summary['positions_seen'])}\n" + f"{self.separator}\n" + f"{self.format_liquidation_orders(summary.get('orders_submitted'))}" + ) + publish_notification_fn(detailed_text=message, compact_text=message) + return StrategyCycleResult( + result="OK", + execution_summary=dict(summary or {}), + ) + finally: + if ib is not None and hasattr(ib, "disconnect"): + ib.disconnect() + + +def build_runtime_broker_adapters( + *, + host_resolver, + ib_port: int, + ib_client_id: int, + connect_timeout_seconds: int, + connect_attempts: int, + connect_retry_delay_seconds: float, + client_id_retry_offset: int, + ensure_event_loop_fn, + connect_ib_fn, + fetch_portfolio_snapshot_fn, + fetch_quote_snapshots_fn, + submit_order_intent_fn, + order_intent_cls=OrderIntent, + application_get_market_prices_fn, + application_check_order_submitted_fn, + application_execute_rebalance_fn, + execute_paper_liquidation_fn, + translator, + strategy_profile: str, + account_group: str, + service_name: str | None, + account_ids: tuple[str, ...], + dry_run_only: bool, + cash_reserve_ratio: float, + rebalance_threshold_ratio: float, + limit_buy_premium: float, + sell_settle_delay_sec: float, + separator: str, + strategy_display_name: str, + sleep_fn, + printer=print, +) -> IBKRRuntimeBrokerAdapters: + return IBKRRuntimeBrokerAdapters( + host_resolver=host_resolver, + ib_port=int(ib_port), + ib_client_id=int(ib_client_id), + connect_timeout_seconds=int(connect_timeout_seconds), + connect_attempts=int(connect_attempts), + connect_retry_delay_seconds=float(connect_retry_delay_seconds), + client_id_retry_offset=int(client_id_retry_offset), + ensure_event_loop_fn=ensure_event_loop_fn, + connect_ib_fn=connect_ib_fn, + fetch_portfolio_snapshot_fn=fetch_portfolio_snapshot_fn, + fetch_quote_snapshots_fn=fetch_quote_snapshots_fn, + submit_order_intent_fn=submit_order_intent_fn, + order_intent_cls=order_intent_cls, + application_get_market_prices_fn=application_get_market_prices_fn, + application_check_order_submitted_fn=application_check_order_submitted_fn, + application_execute_rebalance_fn=application_execute_rebalance_fn, + execute_paper_liquidation_fn=execute_paper_liquidation_fn, + translator=translator, + strategy_profile=str(strategy_profile), + account_group=str(account_group or ""), + service_name=service_name, + account_ids=tuple(account_ids), + dry_run_only=bool(dry_run_only), + cash_reserve_ratio=float(cash_reserve_ratio), + rebalance_threshold_ratio=float(rebalance_threshold_ratio), + limit_buy_premium=float(limit_buy_premium), + sell_settle_delay_sec=float(sell_settle_delay_sec), + separator=str(separator or ""), + strategy_display_name=str(strategy_display_name or ""), + sleep_fn=sleep_fn, + printer=printer, + ) diff --git a/application/runtime_composer.py b/application/runtime_composer.py new file mode 100644 index 0000000..882150e --- /dev/null +++ b/application/runtime_composer.py @@ -0,0 +1,220 @@ +"""Top-level runtime composer for IBKR application wiring.""" + +from __future__ import annotations + +from collections.abc import Callable +from dataclasses import dataclass +from typing import Any + +from application.runtime_dependencies import IBKRRebalanceConfig, IBKRRebalanceRuntime +from application.runtime_notification_adapters import build_runtime_notification_adapters +from application.runtime_reporting_adapters import build_runtime_reporting_adapters +from quant_platform_kit.common.port_adapters import CallablePortfolioPort + + +@dataclass(frozen=True) +class IBKRRuntimeComposer: + service_name: str + strategy_profile: str + strategy_domain: str | None + account_group: str + project_id: str | None + instance_name: str | None + account_ids: tuple[str, ...] + strategy_target_mode: str | None + strategy_artifact_dir: str | None + strategy_display_name: str + strategy_display_name_localized: str + managed_symbols: tuple[str, ...] + signal_source: str + status_icon: str + safe_haven: str + dry_run_only: bool + strategy_config_source: str | None + ib_gateway_host_resolver: Callable[[], str] + ib_gateway_port: int + ib_gateway_mode: str + ib_gateway_ip_mode: str + ib_client_id: int + ib_connect_timeout_seconds: int + feature_snapshot_path: str | None + feature_snapshot_manifest_path: str | None + strategy_config_path: str | None + reconciliation_output_path: str | None + translator: Callable[..., str] + separator: str + send_message: Callable[[str], None] + connect_ib_fn: Callable[[], Any] + build_portfolio_snapshot_fn: Callable[[Any], Any] + compute_signals_fn: Callable[[Any, set[str]], tuple[Any, ...]] + execute_rebalance_fn: Callable[..., Any] + run_id_builder: Callable[[], str] + event_logger: Callable[..., dict[str, Any]] + report_builder: Callable[..., dict[str, Any]] + report_persister: Callable[..., Any] + trace_extractor: Callable[..., str | None] + env_reader: Callable[[str, str], str | None] + printer: Callable[..., Any] = print + notification_builder: Callable[..., Any] = build_runtime_notification_adapters + reporting_builder: Callable[..., Any] = build_runtime_reporting_adapters + + def build_notification_adapters(self): + return self.notification_builder( + send_message=self.send_message, + log_message=lambda message: self.printer(message, flush=True), + ) + + def build_reporting_adapters(self): + return self.reporting_builder( + platform="interactive_brokers", + deploy_target="cloud_run", + service_name=self.service_name, + strategy_profile=self.strategy_profile, + strategy_domain=self.strategy_domain, + account_scope=self.account_group, + account_group=self.account_group, + project_id=self.project_id, + instance_name=self.instance_name, + extra_context_fields={ + "account_ids": list(self.account_ids), + "strategy_target_mode": self.strategy_target_mode, + "strategy_artifact_dir": self.strategy_artifact_dir, + "strategy_display_name": self.strategy_display_name, + "strategy_display_name_localized": self.strategy_display_name_localized, + }, + managed_symbols=self.managed_symbols, + signal_source=self.signal_source, + status_icon=self.status_icon, + safe_haven=self.safe_haven, + strategy_display_name=self.strategy_display_name, + strategy_display_name_localized=self.strategy_display_name_localized, + dry_run=self.dry_run_only, + strategy_config_source=self.strategy_config_source, + ib_gateway_host_resolver=self.ib_gateway_host_resolver, + ib_gateway_port=self.ib_gateway_port, + ib_gateway_mode=self.ib_gateway_mode, + ib_gateway_ip_mode=self.ib_gateway_ip_mode, + ib_client_id=self.ib_client_id, + ib_connect_timeout_seconds=self.ib_connect_timeout_seconds, + feature_snapshot_path=self.feature_snapshot_path, + feature_snapshot_manifest_path=self.feature_snapshot_manifest_path, + strategy_config_path=self.strategy_config_path, + reconciliation_output_path=self.reconciliation_output_path, + report_base_dir=self.env_reader("EXECUTION_REPORT_OUTPUT_DIR", ""), + report_gcs_prefix_uri=self.env_reader("EXECUTION_REPORT_GCS_URI", ""), + run_id_builder=self.run_id_builder, + event_logger=self.event_logger, + report_builder=self.report_builder, + report_persister=self.report_persister, + trace_extractor=self.trace_extractor, + printer=lambda line: self.printer(line, flush=True), + ) + + def build_rebalance_runtime(self): + notification_adapters = self.build_notification_adapters() + return IBKRRebalanceRuntime( + connect_ib=self.connect_ib_fn, + portfolio_port_factory=lambda ib: CallablePortfolioPort( + lambda: self.build_portfolio_snapshot_fn(ib) + ), + compute_signals=self.compute_signals_fn, + execute_rebalance=self.execute_rebalance_fn, + notifications=notification_adapters.notification_port, + ) + + def build_rebalance_config(self): + return IBKRRebalanceConfig( + translator=self.translator, + separator=self.separator, + strategy_display_name=self.strategy_display_name_localized, + reconciliation_output_path=self.reconciliation_output_path, + ) + + +def build_runtime_composer( + *, + service_name: str, + strategy_profile: str, + strategy_domain: str | None, + account_group: str, + project_id: str | None, + instance_name: str | None, + account_ids: tuple[str, ...], + strategy_target_mode: str | None, + strategy_artifact_dir: str | None, + strategy_display_name: str, + strategy_display_name_localized: str, + managed_symbols: tuple[str, ...], + signal_source: str, + status_icon: str, + safe_haven: str, + dry_run_only: bool, + strategy_config_source: str | None, + ib_gateway_host_resolver: Callable[[], str], + ib_gateway_port: int, + ib_gateway_mode: str, + ib_gateway_ip_mode: str, + ib_client_id: int, + ib_connect_timeout_seconds: int, + feature_snapshot_path: str | None, + feature_snapshot_manifest_path: str | None, + strategy_config_path: str | None, + reconciliation_output_path: str | None, + translator: Callable[..., str], + separator: str, + send_message: Callable[[str], None], + connect_ib_fn: Callable[[], Any], + build_portfolio_snapshot_fn: Callable[[Any], Any], + compute_signals_fn: Callable[[Any, set[str]], tuple[Any, ...]], + execute_rebalance_fn: Callable[..., Any], + run_id_builder: Callable[[], str], + event_logger: Callable[..., dict[str, Any]], + report_builder: Callable[..., dict[str, Any]], + report_persister: Callable[..., Any], + trace_extractor: Callable[..., str | None], + env_reader: Callable[[str, str], str | None], + printer: Callable[..., Any] = print, +) -> IBKRRuntimeComposer: + return IBKRRuntimeComposer( + service_name=str(service_name or ""), + strategy_profile=str(strategy_profile), + strategy_domain=strategy_domain, + account_group=str(account_group or ""), + project_id=project_id, + instance_name=instance_name, + account_ids=tuple(account_ids), + strategy_target_mode=strategy_target_mode, + strategy_artifact_dir=strategy_artifact_dir, + strategy_display_name=str(strategy_display_name or ""), + strategy_display_name_localized=str(strategy_display_name_localized or ""), + managed_symbols=tuple(managed_symbols), + signal_source=str(signal_source or ""), + status_icon=str(status_icon or ""), + safe_haven=str(safe_haven or ""), + dry_run_only=bool(dry_run_only), + strategy_config_source=strategy_config_source, + ib_gateway_host_resolver=ib_gateway_host_resolver, + ib_gateway_port=int(ib_gateway_port), + ib_gateway_mode=str(ib_gateway_mode or ""), + ib_gateway_ip_mode=str(ib_gateway_ip_mode or ""), + ib_client_id=int(ib_client_id), + ib_connect_timeout_seconds=int(ib_connect_timeout_seconds), + feature_snapshot_path=feature_snapshot_path, + feature_snapshot_manifest_path=feature_snapshot_manifest_path, + strategy_config_path=strategy_config_path, + reconciliation_output_path=reconciliation_output_path, + translator=translator, + separator=str(separator or ""), + send_message=send_message, + connect_ib_fn=connect_ib_fn, + build_portfolio_snapshot_fn=build_portfolio_snapshot_fn, + compute_signals_fn=compute_signals_fn, + execute_rebalance_fn=execute_rebalance_fn, + run_id_builder=run_id_builder, + event_logger=event_logger, + report_builder=report_builder, + report_persister=report_persister, + trace_extractor=trace_extractor, + env_reader=env_reader, + printer=printer, + ) diff --git a/application/runtime_dependencies.py b/application/runtime_dependencies.py new file mode 100644 index 0000000..bb90ff5 --- /dev/null +++ b/application/runtime_dependencies.py @@ -0,0 +1,27 @@ +"""Runtime dependency bundles for IBKR rebalance orchestration.""" + +from __future__ import annotations + +from collections.abc import Callable +from dataclasses import dataclass +from pathlib import Path +from typing import Any + +from quant_platform_kit.common.ports import NotificationPort, PortfolioPort + + +@dataclass(frozen=True) +class IBKRRebalanceConfig: + translator: Callable[..., str] + separator: str + strategy_display_name: str | None = None + reconciliation_output_path: str | Path | None = None + + +@dataclass(frozen=True) +class IBKRRebalanceRuntime: + connect_ib: Callable[[], Any] + portfolio_port_factory: Callable[[Any], PortfolioPort] + compute_signals: Callable[[Any, set[str]], tuple[Any, ...]] + execute_rebalance: Callable[..., Any] + notifications: NotificationPort diff --git a/application/runtime_notification_adapters.py b/application/runtime_notification_adapters.py new file mode 100644 index 0000000..3d6908b --- /dev/null +++ b/application/runtime_notification_adapters.py @@ -0,0 +1,37 @@ +"""Builder helpers for IBKR runtime notification adapters.""" + +from __future__ import annotations + +from dataclasses import dataclass + +from notifications.events import NotificationPublisher, RenderedNotification +from quant_platform_kit.common.port_adapters import CallableNotificationPort +from quant_platform_kit.common.ports import NotificationPort + + +@dataclass(frozen=True) +class IBKRNotificationAdapters: + notification_port: NotificationPort + cycle_publisher: NotificationPublisher + + def publish_cycle_notification(self, *, detailed_text: str, compact_text: str) -> None: + self.cycle_publisher.publish( + RenderedNotification( + detailed_text=detailed_text, + compact_text=compact_text, + ) + ) + + +def build_runtime_notification_adapters( + *, + send_message, + log_message=None, +) -> IBKRNotificationAdapters: + return IBKRNotificationAdapters( + notification_port=CallableNotificationPort(send_message), + cycle_publisher=NotificationPublisher( + log_message=log_message or (lambda message: print(message, flush=True)), + send_message=send_message, + ), + ) diff --git a/application/runtime_reporting_adapters.py b/application/runtime_reporting_adapters.py new file mode 100644 index 0000000..ab54667 --- /dev/null +++ b/application/runtime_reporting_adapters.py @@ -0,0 +1,227 @@ +"""Builder helpers for IBKR runtime reporting and structured logging.""" + +from __future__ import annotations + +from collections.abc import Callable, Mapping +from dataclasses import dataclass, field +from datetime import datetime, timezone +from typing import Any + +from runtime_logging import RuntimeLogContext + + +def _utcnow() -> datetime: + return datetime.now(timezone.utc) + + +@dataclass(frozen=True) +class IBKRRuntimeReportingAdapters: + platform: str + deploy_target: str + service_name: str + strategy_profile: str + strategy_domain: str | None + account_scope: str | None + account_group: str | None + project_id: str | None + instance_name: str | None + extra_context_fields: Mapping[str, Any] = field(default_factory=dict) + managed_symbols: tuple[str, ...] = () + signal_source: str = "" + status_icon: str = "" + safe_haven: str = "" + strategy_display_name: str = "" + strategy_display_name_localized: str = "" + dry_run: bool = False + strategy_config_source: str | None = None + ib_gateway_host_resolver: Callable[[], str] | None = None + ib_gateway_port: int = 0 + ib_gateway_mode: str = "" + ib_gateway_ip_mode: str = "" + ib_client_id: int = 0 + ib_connect_timeout_seconds: int = 0 + feature_snapshot_path: str | None = None + feature_snapshot_manifest_path: str | None = None + strategy_config_path: str | None = None + reconciliation_output_path: str | None = None + report_base_dir: str | None = None + report_gcs_prefix_uri: str | None = None + run_id_builder: Callable[[], str] | None = None + event_logger: Callable[..., dict[str, Any]] | None = None + report_builder: Callable[..., dict[str, Any]] | None = None + report_persister: Callable[..., Any] | None = None + trace_extractor: Callable[..., str | None] | None = None + printer: Callable[..., Any] = print + clock: Callable[[], datetime] = _utcnow + + def __post_init__(self) -> None: + required = { + "ib_gateway_host_resolver": self.ib_gateway_host_resolver, + "run_id_builder": self.run_id_builder, + "event_logger": self.event_logger, + "report_builder": self.report_builder, + "report_persister": self.report_persister, + "trace_extractor": self.trace_extractor, + } + missing = [name for name, value in required.items() if value is None] + if missing: + raise ValueError(f"Missing reporting adapter dependencies: {', '.join(missing)}") + + def build_log_context(self, *, trace_header: str | None = None) -> RuntimeLogContext: + return RuntimeLogContext( + platform=self.platform, + deploy_target=self.deploy_target, + service_name=self.service_name, + strategy_profile=self.strategy_profile, + account_scope=self.account_scope, + account_group=self.account_group, + project_id=self.project_id, + instance_name=self.instance_name, + extra_fields=dict(self.extra_context_fields), + ).with_run( + self.run_id_builder(), + trace=self.trace_extractor(self.project_id, trace_header), + ) + + def build_report(self, log_context: RuntimeLogContext) -> dict[str, Any]: + return self.report_builder( + platform=log_context.platform, + deploy_target=log_context.deploy_target, + service_name=log_context.service_name, + strategy_profile=self.strategy_profile, + strategy_domain=self.strategy_domain, + account_scope=log_context.account_scope, + account_group=log_context.account_group, + run_id=log_context.run_id, + run_source="cloud_run", + dry_run=self.dry_run, + started_at=self.clock(), + summary={ + "account_ids": list(self.extra_context_fields.get("account_ids") or ()), + "managed_symbols": list(self.managed_symbols), + "signal_source": self.signal_source, + "status_icon": self.status_icon, + "safe_haven": self.safe_haven, + "strategy_display_name": self.strategy_display_name, + "strategy_display_name_localized": self.strategy_display_name_localized, + }, + diagnostics={ + "strategy_config_source": self.strategy_config_source, + "ib_gateway_host": self.ib_gateway_host_resolver(), + "ib_gateway_port": self.ib_gateway_port, + "ib_gateway_mode": self.ib_gateway_mode, + "ib_gateway_ip_mode": self.ib_gateway_ip_mode, + "ib_client_id": self.ib_client_id, + "ib_connect_timeout_seconds": self.ib_connect_timeout_seconds, + }, + artifacts={ + "feature_snapshot_path": self.feature_snapshot_path, + "feature_snapshot_manifest_path": self.feature_snapshot_manifest_path, + "strategy_config_path": self.strategy_config_path, + "reconciliation_output_path": self.reconciliation_output_path, + }, + ) + + def start_request_run(self, *, trace_header: str | None = None) -> tuple[RuntimeLogContext, dict[str, Any]]: + log_context = self.build_log_context(trace_header=trace_header) + return log_context, self.build_report(log_context) + + def log_event(self, log_context: RuntimeLogContext, event: str, **fields: Any) -> dict[str, Any]: + return self.event_logger( + log_context, + event, + printer=self.printer, + **fields, + ) + + def persist_execution_report(self, report: dict[str, Any]) -> str | None: + persisted = self.report_persister( + report, + base_dir=self.report_base_dir, + gcs_prefix_uri=self.report_gcs_prefix_uri, + gcp_project_id=self.project_id, + ) + if isinstance(persisted, str): + return persisted + return getattr(persisted, "gcs_uri", None) or getattr(persisted, "local_path", None) + + +def build_runtime_reporting_adapters( + *, + platform: str, + deploy_target: str, + service_name: str, + strategy_profile: str, + strategy_domain: str | None, + account_scope: str | None, + account_group: str | None, + project_id: str | None, + instance_name: str | None, + extra_context_fields: Mapping[str, Any] | None = None, + managed_symbols: tuple[str, ...], + signal_source: str, + status_icon: str, + safe_haven: str, + strategy_display_name: str, + strategy_display_name_localized: str, + dry_run: bool, + strategy_config_source: str | None, + ib_gateway_host_resolver: Callable[[], str], + ib_gateway_port: int, + ib_gateway_mode: str, + ib_gateway_ip_mode: str, + ib_client_id: int, + ib_connect_timeout_seconds: int, + feature_snapshot_path: str | None, + feature_snapshot_manifest_path: str | None, + strategy_config_path: str | None, + reconciliation_output_path: str | None, + report_base_dir: str | None, + report_gcs_prefix_uri: str | None, + run_id_builder: Callable[[], str], + event_logger: Callable[..., dict[str, Any]], + report_builder: Callable[..., dict[str, Any]], + report_persister: Callable[..., Any], + trace_extractor: Callable[..., str | None], + printer: Callable[..., Any] = print, + clock: Callable[[], datetime] = _utcnow, +) -> IBKRRuntimeReportingAdapters: + return IBKRRuntimeReportingAdapters( + platform=platform, + deploy_target=deploy_target, + service_name=service_name, + strategy_profile=strategy_profile, + strategy_domain=strategy_domain, + account_scope=account_scope, + account_group=account_group, + project_id=project_id, + instance_name=instance_name, + extra_context_fields=dict(extra_context_fields or {}), + managed_symbols=tuple(managed_symbols), + signal_source=str(signal_source or ""), + status_icon=str(status_icon or ""), + safe_haven=str(safe_haven or ""), + strategy_display_name=str(strategy_display_name or ""), + strategy_display_name_localized=str(strategy_display_name_localized or ""), + dry_run=bool(dry_run), + strategy_config_source=strategy_config_source, + ib_gateway_host_resolver=ib_gateway_host_resolver, + ib_gateway_port=int(ib_gateway_port), + ib_gateway_mode=str(ib_gateway_mode or ""), + ib_gateway_ip_mode=str(ib_gateway_ip_mode or ""), + ib_client_id=int(ib_client_id), + ib_connect_timeout_seconds=int(ib_connect_timeout_seconds), + feature_snapshot_path=feature_snapshot_path, + feature_snapshot_manifest_path=feature_snapshot_manifest_path, + strategy_config_path=strategy_config_path, + reconciliation_output_path=reconciliation_output_path, + report_base_dir=report_base_dir, + report_gcs_prefix_uri=report_gcs_prefix_uri, + run_id_builder=run_id_builder, + event_logger=event_logger, + report_builder=report_builder, + report_persister=report_persister, + trace_extractor=trace_extractor, + printer=printer, + clock=clock, + ) diff --git a/application/runtime_strategy_adapters.py b/application/runtime_strategy_adapters.py new file mode 100644 index 0000000..cf42805 --- /dev/null +++ b/application/runtime_strategy_adapters.py @@ -0,0 +1,81 @@ +"""Builder helpers for IBKR strategy evaluation adapters.""" + +from __future__ import annotations + +from dataclasses import dataclass +from typing import Any + +import pandas as pd + + +@dataclass(frozen=True) +class IBKRRuntimeStrategyAdapters: + strategy_runtime: Any + strategy_profile: str + translator: Any + pacing_sec: float + resolve_run_as_of_date_fn: Any + fetch_historical_price_series_fn: Any + fetch_historical_price_candles_fn: Any + map_strategy_decision_fn: Any + + def get_historical_close(self, ib, symbol, duration="2 Y", bar_size="1 day"): + series = self.fetch_historical_price_series_fn( + ib, + symbol, + duration=duration, + bar_size=bar_size, + ) + if not series.points: + return pd.Series(dtype=float) + return pd.Series( + data=[point.close for point in series.points], + index=pd.to_datetime([point.as_of for point in series.points]), + ) + + def get_historical_candles(self, ib, symbol, duration="2 Y", bar_size="1 day"): + return self.fetch_historical_price_candles_fn( + ib, + symbol, + duration=duration, + bar_size=bar_size, + ) + + def compute_signals(self, ib, current_holdings): + evaluation = self.strategy_runtime.evaluate( + ib=ib, + current_holdings=current_holdings, + historical_close_loader=self.get_historical_close, + historical_candle_loader=self.get_historical_candles, + run_as_of=self.resolve_run_as_of_date_fn(), + translator=self.translator, + pacing_sec=self.pacing_sec, + ) + return self.map_strategy_decision_fn( + evaluation.decision, + strategy_profile=self.strategy_profile, + runtime_metadata=evaluation.metadata, + ) + + +def build_runtime_strategy_adapters( + *, + strategy_runtime: Any, + strategy_profile: str, + translator, + pacing_sec: float, + resolve_run_as_of_date_fn, + fetch_historical_price_series_fn, + fetch_historical_price_candles_fn, + map_strategy_decision_fn, +) -> IBKRRuntimeStrategyAdapters: + return IBKRRuntimeStrategyAdapters( + strategy_runtime=strategy_runtime, + strategy_profile=str(strategy_profile), + translator=translator, + pacing_sec=float(pacing_sec), + resolve_run_as_of_date_fn=resolve_run_as_of_date_fn, + fetch_historical_price_series_fn=fetch_historical_price_series_fn, + fetch_historical_price_candles_fn=fetch_historical_price_candles_fn, + map_strategy_decision_fn=map_strategy_decision_fn, + ) diff --git a/main.py b/main.py index 040bc07..b901afc 100644 --- a/main.py +++ b/main.py @@ -1,24 +1,30 @@ """IBKR strategy runner for shared us_equity strategy profiles.""" + import os import threading import time import traceback -from datetime import datetime, timezone +from datetime import datetime from zoneinfo import ZoneInfo -import requests +import google.auth import pandas as pd +import requests from flask import Flask, request -import google.auth try: from google.cloud import compute_v1 except ImportError: compute_v1 = None -from notifications.events import NotificationPublisher, RenderedNotification +from application.cycle_result import coerce_strategy_cycle_result +from application.runtime_broker_adapters import build_runtime_broker_adapters +from application.runtime_composer import build_runtime_composer +from application.runtime_strategy_adapters import build_runtime_strategy_adapters +from application.rebalance_service import run_strategy_core as run_rebalance_cycle +from decision_mapper import map_strategy_decision +from entrypoints.cloud_run import is_market_open_today from notifications.telegram import build_strategy_display_name, build_translator, send_telegram_message -from quant_platform_kit.common.models import OrderIntent from quant_platform_kit.common.runtime_reports import ( append_runtime_report_error, build_runtime_report_base, @@ -40,19 +46,8 @@ get_market_prices as application_get_market_prices, ) from application.paper_liquidation_service import execute_paper_liquidation -from application.rebalance_service import run_strategy_core as run_rebalance_cycle -from decision_mapper import map_strategy_decision -from entrypoints.cloud_run import is_market_open_today -from runtime_logging import ( - RuntimeLogContext, - build_run_id, - emit_runtime_log, - extract_cloud_trace, -) -from runtime_config_support import ( - load_platform_runtime_settings, - resolve_ib_gateway_ip_mode, -) +from runtime_logging import RuntimeLogContext, build_run_id, emit_runtime_log, extract_cloud_trace +from runtime_config_support import load_platform_runtime_settings, resolve_ib_gateway_ip_mode from strategy_runtime import load_strategy_runtime app = Flask(__name__) @@ -61,9 +56,6 @@ STRATEGY_RUN_LOCK = threading.Lock() -# --------------------------------------------------------------------------- -# GCE instance resolver: find IB Gateway by instance name instead of IP -# --------------------------------------------------------------------------- def get_project_id(): try: _, project_id = google.auth.default() @@ -83,7 +75,6 @@ def get_ib_gateway_ip_mode(): def resolve_gce_instance_ip(instance_name, zone): - """Resolve GCE instance IP by name via Compute API.""" if not compute_v1: print(f"google-cloud-compute not installed, using {instance_name} as host directly", flush=True) return instance_name @@ -110,23 +101,15 @@ def resolve_gce_instance_ip(instance_name, zone): if ip: print(f"Resolved {instance_name} → {ip} ({label}, mode={ip_mode})", flush=True) return ip - except Exception as e: - print(f"GCE resolve failed for {instance_name}: {e}, using as hostname", flush=True) + except Exception as exc: + print(f"GCE resolve failed for {instance_name}: {exc}, using as hostname", flush=True) return instance_name def get_ib_host(): - """ - Resolve IB Gateway host lazily. - - Read IB_GATEWAY_INSTANCE_NAME only - - If IB_GATEWAY_ZONE is set: resolve instance name via Compute API - - If IB_GATEWAY_ZONE is not set: use the configured instance name directly - """ global IB_HOST - if IB_HOST: return IB_HOST - host = RUNTIME_SETTINGS.ib_gateway_instance_name zone = RUNTIME_SETTINGS.ib_gateway_zone if zone: @@ -140,8 +123,7 @@ def get_ib_gateway_mode(): def get_ib_port(): - mode = get_ib_gateway_mode() - return 4002 if mode == "paper" else 4001 + return 4002 if get_ib_gateway_mode() == "paper" else 4001 def get_ib_connect_timeout_seconds(): @@ -149,16 +131,10 @@ def get_ib_connect_timeout_seconds(): try: timeout_seconds = int(raw_value) except (TypeError, ValueError): - print( - f"Invalid IBKR_CONNECT_TIMEOUT_SECONDS={raw_value!r}; using 60", - flush=True, - ) + print(f"Invalid IBKR_CONNECT_TIMEOUT_SECONDS={raw_value!r}; using 60", flush=True) return 60 if timeout_seconds <= 0: - print( - f"Invalid IBKR_CONNECT_TIMEOUT_SECONDS={raw_value!r}; using 60", - flush=True, - ) + print(f"Invalid IBKR_CONNECT_TIMEOUT_SECONDS={raw_value!r}; using 60", flush=True) return 60 return timeout_seconds @@ -193,9 +169,6 @@ def _env_flag(name: str) -> bool: return str(os.getenv(name) or "").strip().lower() in {"1", "true", "yes", "on"} -# --------------------------------------------------------------------------- -# Config -# --------------------------------------------------------------------------- RUNTIME_SETTINGS = load_platform_runtime_settings(project_id_resolver=get_project_id) IB_HOST = None IB_PORT = get_ib_port() @@ -250,15 +223,10 @@ def _env_flag(name: str) -> bool: NOTIFY_LANG = RUNTIME_SETTINGS.notify_lang CASH_RESERVE_RATIO = STRATEGY_RUNTIME.cash_reserve_ratio -REBALANCE_THRESHOLD_RATIO = 0.02 # 2% of equity to trigger trades +REBALANCE_THRESHOLD_RATIO = 0.02 LIMIT_BUY_PREMIUM = 1.005 - -# Execution SELL_SETTLE_DELAY_SEC = 3 - -# IBKR pacing: delay between historical data requests HIST_DATA_PACING_SEC = 0.5 - SEPARATOR = "━━━━━━━━━━━━━━━━━━" @@ -271,7 +239,6 @@ def t(key, **kwargs): fallback_name=STRATEGY_DISPLAY_NAME, ) - RUNTIME_LOG_CONTEXT = RuntimeLogContext( platform="interactive_brokers", deploy_target="cloud_run", @@ -291,7 +258,110 @@ def t(key, **kwargs): "ib_client_id_retry_offset": IB_CLIENT_ID_RETRY_OFFSET, }, ) -LAST_CYCLE_DETAILS: dict[str, object] = {} + + +def resolve_reporting_managed_symbols() -> tuple[str, ...]: + configured_managed_symbols = STRATEGY_RUNTIME_CONFIG.get("managed_symbols") + fallback_managed_symbols = tuple(dict.fromkeys([*RANKING_POOL, SAFE_HAVEN])) if RANKING_POOL else (SAFE_HAVEN,) + return tuple( + str(symbol) + for symbol in (configured_managed_symbols or fallback_managed_symbols) + if str(symbol or "").strip() + ) + + +def build_strategy_adapters(): + return build_runtime_strategy_adapters( + strategy_runtime=STRATEGY_RUNTIME, + strategy_profile=STRATEGY_PROFILE, + translator=t, + pacing_sec=HIST_DATA_PACING_SEC, + resolve_run_as_of_date_fn=resolve_run_as_of_date, + fetch_historical_price_series_fn=fetch_historical_price_series, + fetch_historical_price_candles_fn=fetch_historical_price_candles, + map_strategy_decision_fn=map_strategy_decision, + ) + + +def build_broker_adapters(): + return build_runtime_broker_adapters( + host_resolver=get_ib_host, + ib_port=IB_PORT, + ib_client_id=IB_CLIENT_ID, + connect_timeout_seconds=IB_CONNECT_TIMEOUT_SECONDS, + connect_attempts=IB_CONNECT_ATTEMPTS, + connect_retry_delay_seconds=IB_CONNECT_RETRY_DELAY_SECONDS, + client_id_retry_offset=IB_CLIENT_ID_RETRY_OFFSET, + ensure_event_loop_fn=ensure_event_loop, + connect_ib_fn=ibkr_connect_ib, + fetch_portfolio_snapshot_fn=fetch_portfolio_snapshot, + fetch_quote_snapshots_fn=fetch_quote_snapshots, + submit_order_intent_fn=submit_order_intent, + application_get_market_prices_fn=application_get_market_prices, + application_check_order_submitted_fn=application_check_order_submitted, + application_execute_rebalance_fn=application_execute_rebalance, + execute_paper_liquidation_fn=execute_paper_liquidation, + translator=t, + strategy_profile=STRATEGY_PROFILE, + account_group=ACCOUNT_GROUP, + service_name=SERVICE_NAME, + account_ids=tuple(ACCOUNT_IDS), + dry_run_only=RUNTIME_SETTINGS.dry_run_only, + cash_reserve_ratio=CASH_RESERVE_RATIO, + rebalance_threshold_ratio=REBALANCE_THRESHOLD_RATIO, + limit_buy_premium=LIMIT_BUY_PREMIUM, + sell_settle_delay_sec=SELL_SETTLE_DELAY_SEC, + separator=SEPARATOR, + strategy_display_name=strategy_display_name, + sleep_fn=time.sleep, + printer=print, + ) + + +def build_composer(): + return build_runtime_composer( + service_name=SERVICE_NAME or os.getenv("K_SERVICE", "interactive-brokers-platform"), + strategy_profile=STRATEGY_PROFILE, + strategy_domain=RUNTIME_SETTINGS.strategy_domain, + account_group=ACCOUNT_GROUP, + project_id=PROJECT_ID, + instance_name=RUNTIME_SETTINGS.ib_gateway_instance_name, + account_ids=tuple(ACCOUNT_IDS), + strategy_target_mode=RUNTIME_SETTINGS.strategy_target_mode, + strategy_artifact_dir=RUNTIME_SETTINGS.strategy_artifact_dir, + strategy_display_name=STRATEGY_DISPLAY_NAME, + strategy_display_name_localized=strategy_display_name, + managed_symbols=resolve_reporting_managed_symbols(), + signal_source=STRATEGY_SIGNAL_SOURCE, + status_icon=STRATEGY_STATUS_ICON, + safe_haven=SAFE_HAVEN, + dry_run_only=RUNTIME_SETTINGS.dry_run_only, + strategy_config_source=FEATURE_RUNTIME_CONFIG_SOURCE, + ib_gateway_host_resolver=get_ib_host, + ib_gateway_port=IB_PORT, + ib_gateway_mode=RUNTIME_SETTINGS.ib_gateway_mode, + ib_gateway_ip_mode=RUNTIME_SETTINGS.ib_gateway_ip_mode, + ib_client_id=IB_CLIENT_ID, + ib_connect_timeout_seconds=IB_CONNECT_TIMEOUT_SECONDS, + feature_snapshot_path=FEATURE_SNAPSHOT_PATH, + feature_snapshot_manifest_path=FEATURE_SNAPSHOT_MANIFEST_PATH, + strategy_config_path=FEATURE_RUNTIME_CONFIG_PATH, + reconciliation_output_path=RECONCILIATION_OUTPUT_PATH, + translator=t, + separator=SEPARATOR, + send_message=send_tg_message, + connect_ib_fn=connect_ib, + build_portfolio_snapshot_fn=build_portfolio_snapshot, + compute_signals_fn=compute_signals, + execute_rebalance_fn=execute_rebalance, + run_id_builder=build_run_id, + event_logger=emit_runtime_log, + report_builder=build_runtime_report_base, + report_persister=persist_runtime_report, + trace_extractor=extract_cloud_trace, + env_reader=os.getenv, + printer=print, + ) def send_tg_message(message): @@ -304,128 +374,31 @@ def send_tg_message(message): def publish_notification(*, detailed_text, compact_text): - publisher = NotificationPublisher( - log_message=lambda message: print(message, flush=True), - send_message=send_tg_message, - ) - publisher.publish( - RenderedNotification( - detailed_text=detailed_text, - compact_text=compact_text, - ) + build_composer().build_notification_adapters().publish_cycle_notification( + detailed_text=detailed_text, + compact_text=compact_text, ) def connect_ib(): - host = get_ib_host() - last_error = None - for attempt in range(1, IB_CONNECT_ATTEMPTS + 1): - client_id = IB_CLIENT_ID + ((attempt - 1) * IB_CLIENT_ID_RETRY_OFFSET) - print( - "Connecting to IB gateway " - f"{host}:{IB_PORT} " - f"(mode={RUNTIME_SETTINGS.ib_gateway_mode}, " - f"client_id={client_id}, " - f"attempt={attempt}/{IB_CONNECT_ATTEMPTS}, " - f"timeout={IB_CONNECT_TIMEOUT_SECONDS}s)", - flush=True, - ) - try: - return ibkr_connect_ib( - host, - IB_PORT, - client_id, - timeout=IB_CONNECT_TIMEOUT_SECONDS, - ) - except (ConnectionError, TimeoutError, OSError) as exc: - last_error = exc - print( - "IB gateway connection attempt failed " - f"(attempt={attempt}/{IB_CONNECT_ATTEMPTS}, " - f"client_id={client_id}, " - f"error_type={type(exc).__name__}, " - f"error={exc})", - flush=True, - ) - if attempt < IB_CONNECT_ATTEMPTS and IB_CONNECT_RETRY_DELAY_SECONDS > 0: - time.sleep(IB_CONNECT_RETRY_DELAY_SECONDS) - - raise last_error + return build_broker_adapters().connect_ib() def log_runtime_event(log_context, event, **fields): - return emit_runtime_log( - log_context, - event, - printer=lambda line: print(line, flush=True), - **fields, - ) + return build_composer().build_reporting_adapters().log_event(log_context, event, **fields) def build_execution_report(log_context): - configured_managed_symbols = STRATEGY_RUNTIME_CONFIG.get("managed_symbols") - fallback_managed_symbols = tuple(dict.fromkeys([*RANKING_POOL, SAFE_HAVEN])) if RANKING_POOL else (SAFE_HAVEN,) - managed_symbols = tuple( - str(symbol) - for symbol in (configured_managed_symbols or fallback_managed_symbols) - if str(symbol or "").strip() - ) - return build_runtime_report_base( - platform=log_context.platform, - deploy_target=log_context.deploy_target, - service_name=log_context.service_name, - strategy_profile=STRATEGY_PROFILE, - strategy_domain=RUNTIME_SETTINGS.strategy_domain, - account_scope=log_context.account_scope, - account_group=log_context.account_group, - run_id=log_context.run_id, - run_source="cloud_run", - dry_run=RUNTIME_SETTINGS.dry_run_only, - started_at=datetime.now(timezone.utc), - summary={ - "account_ids": list(ACCOUNT_IDS), - "managed_symbols": list(managed_symbols), - "signal_source": STRATEGY_SIGNAL_SOURCE, - "status_icon": STRATEGY_STATUS_ICON, - "safe_haven": SAFE_HAVEN, - "strategy_display_name": STRATEGY_DISPLAY_NAME, - "strategy_display_name_localized": strategy_display_name, - }, - diagnostics={ - "strategy_config_source": FEATURE_RUNTIME_CONFIG_SOURCE, - "ib_gateway_host": get_ib_host(), - "ib_gateway_port": IB_PORT, - "ib_gateway_mode": RUNTIME_SETTINGS.ib_gateway_mode, - "ib_gateway_ip_mode": RUNTIME_SETTINGS.ib_gateway_ip_mode, - "ib_client_id": IB_CLIENT_ID, - "ib_connect_timeout_seconds": IB_CONNECT_TIMEOUT_SECONDS, - }, - artifacts={ - "feature_snapshot_path": FEATURE_SNAPSHOT_PATH, - "feature_snapshot_manifest_path": FEATURE_SNAPSHOT_MANIFEST_PATH, - "strategy_config_path": FEATURE_RUNTIME_CONFIG_PATH, - "reconciliation_output_path": RECONCILIATION_OUTPUT_PATH, - }, - ) + return build_composer().build_reporting_adapters().build_report(log_context) def persist_execution_report(report): - persisted = persist_runtime_report( - report, - base_dir=os.getenv("EXECUTION_REPORT_OUTPUT_DIR"), - gcs_prefix_uri=os.getenv("EXECUTION_REPORT_GCS_URI"), - gcp_project_id=PROJECT_ID, - ) - return persisted.gcs_uri or persisted.local_path + return build_composer().build_reporting_adapters().persist_execution_report(report) def build_request_log_context(): - return RUNTIME_LOG_CONTEXT.with_run( - build_run_id(), - trace=extract_cloud_trace( - PROJECT_ID, - request.headers.get("X-Cloud-Trace-Context"), - ), + return build_composer().build_reporting_adapters().build_log_context( + trace_header=request.headers.get("X-Cloud-Trace-Context"), ) @@ -437,24 +410,16 @@ def resolve_run_as_of_date() -> pd.Timestamp: def get_historical_close(ib, symbol, duration="2 Y", bar_size="1 day"): - """Fetch daily close prices from IBKR via QuantPlatformKit.""" - series = fetch_historical_price_series( + return build_strategy_adapters().get_historical_close( ib, symbol, duration=duration, bar_size=bar_size, ) - if not series.points: - return pd.Series(dtype=float) - return pd.Series( - data=[point.close for point in series.points], - index=pd.to_datetime([point.as_of for point in series.points]), - ) def get_historical_candles(ib, symbol, duration="2 Y", bar_size="1 day"): - """Fetch daily OHLC candles from IBKR via QuantPlatformKit.""" - return fetch_historical_price_candles( + return build_strategy_adapters().get_historical_candles( ib, symbol, duration=duration, @@ -462,57 +427,27 @@ def get_historical_candles(ib, symbol, duration="2 Y", bar_size="1 day"): ) -# --------------------------------------------------------------------------- -# Strategy logic -# --------------------------------------------------------------------------- def compute_signals(ib, current_holdings): - evaluation = STRATEGY_RUNTIME.evaluate( - ib=ib, - current_holdings=current_holdings, - historical_close_loader=get_historical_close, - historical_candle_loader=get_historical_candles, - run_as_of=resolve_run_as_of_date(), - translator=t, - pacing_sec=HIST_DATA_PACING_SEC, - ) - return map_strategy_decision( - evaluation.decision, - strategy_profile=STRATEGY_PROFILE, - runtime_metadata=evaluation.metadata, - ) + return build_strategy_adapters().compute_signals(ib, current_holdings) -# --------------------------------------------------------------------------- -# Portfolio execution -# --------------------------------------------------------------------------- def get_current_portfolio(ib): - """Get current positions and account values.""" - snapshot = fetch_portfolio_snapshot(ib) - positions = {} - for position in snapshot.positions: - positions[position.symbol] = { - 'quantity': int(position.quantity), - 'avg_cost': float(position.average_cost or 0.0), - } + return build_broker_adapters().get_current_portfolio(ib) - account_values = { - 'equity': snapshot.total_equity, - 'buying_power': snapshot.buying_power or 0.0, - } - return positions, account_values +def build_portfolio_snapshot(ib): + return build_broker_adapters().build_portfolio_snapshot( + ib, + get_current_portfolio_fallback=get_current_portfolio, + ) def get_market_prices(ib, symbols): - return application_get_market_prices( - ib, - symbols, - fetch_quote_snapshots=fetch_quote_snapshots, - ) + return build_broker_adapters().get_market_prices(ib, symbols) def check_order_submitted(report): - return application_check_order_submitted(report, translator=t) + return build_broker_adapters().check_order_submitted(report) def execute_rebalance( @@ -524,113 +459,45 @@ def execute_rebalance( strategy_symbols=None, signal_metadata=None, ): - return application_execute_rebalance( + return build_broker_adapters().execute_rebalance( ib, target_weights, positions, account_values, - fetch_quote_snapshots=fetch_quote_snapshots, - submit_order_intent=submit_order_intent, - order_intent_cls=OrderIntent, - translator=t, strategy_symbols=strategy_symbols, - signal_metadata=signal_metadata or {}, - strategy_profile=STRATEGY_PROFILE, - account_group=ACCOUNT_GROUP, - service_name=SERVICE_NAME, - account_ids=ACCOUNT_IDS, - dry_run_only=RUNTIME_SETTINGS.dry_run_only, - cash_reserve_ratio=CASH_RESERVE_RATIO, - rebalance_threshold_ratio=REBALANCE_THRESHOLD_RATIO, - limit_buy_premium=LIMIT_BUY_PREMIUM, - sell_settle_delay_sec=SELL_SETTLE_DELAY_SEC, - return_summary=True, + signal_metadata=signal_metadata, ) def _format_liquidation_orders(orders) -> str: - preview = [] - for order in orders or (): - symbol = str(order.get("symbol") or "").strip().upper() - side = str(order.get("side") or "").strip().lower() - quantity = float(order.get("quantity") or 0.0) - status = str(order.get("status") or "").strip() - if symbol: - preview.append(f"{symbol} {side} {quantity:g} {status}".strip()) - return ", ".join(preview) if preview else t("no_trades") + return build_broker_adapters().format_liquidation_orders(orders) def run_paper_liquidation_cycle(): if RUNTIME_SETTINGS.ib_gateway_mode != "paper": raise RuntimeError("IBKR_PAPER_LIQUIDATE_ONLY is only allowed when ib_gateway_mode=paper") - - ib = connect_ib() - try: - positions, _account_values = get_current_portfolio(ib) - if not positions: - print("paper_liquidation_positions_empty_retry", flush=True) - time.sleep(2.0) - positions, _account_values = get_current_portfolio(ib) - summary = execute_paper_liquidation( - ib, - positions, - submit_order_intent=submit_order_intent, - order_intent_cls=OrderIntent, - dry_run_only=RUNTIME_SETTINGS.dry_run_only, - ) - message = ( - f"{t('rebalance_title')}\n" - f"{t('strategy_label', name=strategy_display_name)}\n" - f"{t('paper_liquidation_only')}\n" - f"{t('paper_liquidation_status', mode=summary['mode'], status=summary['execution_status'])}\n" - f"{t('paper_liquidation_positions_seen', count=summary['positions_seen'])}\n" - f"{SEPARATOR}\n" - f"{_format_liquidation_orders(summary.get('orders_submitted'))}" - ) - publish_notification(detailed_text=message, compact_text=message) - global LAST_CYCLE_DETAILS - LAST_CYCLE_DETAILS = {"execution_summary": summary} - return "OK" - finally: - if ib is not None and hasattr(ib, "disconnect"): - ib.disconnect() + return build_broker_adapters().run_paper_liquidation_cycle( + connect_ib_fn=connect_ib, + get_current_portfolio_fn=get_current_portfolio, + publish_notification_fn=publish_notification, + ) -# --------------------------------------------------------------------------- -# Main strategy runner -# --------------------------------------------------------------------------- def run_strategy_core(): - global LAST_CYCLE_DETAILS if PAPER_LIQUIDATE_ONLY: return run_paper_liquidation_cycle() - - cycle_details: dict[str, object] = {} - result = run_rebalance_cycle( - connect_ib=connect_ib, - get_current_portfolio=get_current_portfolio, - compute_signals=compute_signals, - execute_rebalance=execute_rebalance, - send_tg_message=send_tg_message, - translator=t, - separator=SEPARATOR, - strategy_display_name=strategy_display_name, - reconciliation_output_path=RECONCILIATION_OUTPUT_PATH, - result_hook=lambda payload: cycle_details.update(payload or {}), + composer = build_composer() + return run_rebalance_cycle( + runtime=composer.build_rebalance_runtime(), + config=composer.build_rebalance_config(), ) - LAST_CYCLE_DETAILS = cycle_details - return result -# --------------------------------------------------------------------------- -# Flask routes -# --------------------------------------------------------------------------- @app.route("/", methods=["POST", "GET"]) def handle_request(): if request.method == "GET": return "OK - use POST to execute strategy", 200 - global LAST_CYCLE_DETAILS - LAST_CYCLE_DETAILS = {} log_context = build_request_log_context() report = build_execution_report(log_context) lock_acquired = STRATEGY_RUN_LOCK.acquire(blocking=False) @@ -671,19 +538,26 @@ def handle_request(): "strategy_cycle_started", message="Starting strategy execution", ) - result = run_strategy_core() - cycle_details = dict(LAST_CYCLE_DETAILS or {}) - execution_summary = dict(cycle_details.get("execution_summary") or {}) - reconciliation_record = dict(cycle_details.get("reconciliation_record") or {}) + cycle_result = coerce_strategy_cycle_result(run_strategy_core()) + execution_summary = dict(cycle_result.execution_summary or {}) + reconciliation_record = dict(cycle_result.reconciliation_record or {}) finalize_runtime_report( report, status="ok", summary={ - "result": result, + "result": cycle_result.result, "execution_status": execution_summary.get("execution_status") or reconciliation_record.get("execution_status"), "no_op_reason": execution_summary.get("no_op_reason") or reconciliation_record.get("no_op_reason"), - "orders_submitted_count": len(execution_summary.get("orders_submitted") or reconciliation_record.get("orders_submitted") or ()), - "orders_skipped_count": len(execution_summary.get("orders_skipped") or reconciliation_record.get("orders_skipped") or ()), + "orders_submitted_count": len( + execution_summary.get("orders_submitted") + or reconciliation_record.get("orders_submitted") + or () + ), + "orders_skipped_count": len( + execution_summary.get("orders_skipped") + or reconciliation_record.get("orders_skipped") + or () + ), "snapshot_price_fallback_used": bool( execution_summary.get("snapshot_price_fallback_used") or reconciliation_record.get("snapshot_price_fallback_used") @@ -695,23 +569,23 @@ def handle_request(): ), }, diagnostics={ - "result": result, + "result": cycle_result.result, "price_source_mode": execution_summary.get("price_source_mode") or reconciliation_record.get("price_source_mode"), "snapshot_price_fallback_symbols": execution_summary.get("snapshot_price_fallback_symbols") or reconciliation_record.get("snapshot_price_fallback_symbols") or [], }, artifacts={ - "reconciliation_record_path": cycle_details.get("reconciliation_record_path"), + "reconciliation_record_path": cycle_result.reconciliation_record_path, }, ) log_runtime_event( log_context, "strategy_cycle_completed", message="Strategy execution completed", - result=result, + result=cycle_result.result, ) - return result, 200 + return cycle_result.result, 200 except Exception as exc: append_runtime_report_error( report, @@ -747,4 +621,4 @@ def health(): if __name__ == "__main__": - app.run(host='0.0.0.0', port=int(os.environ.get("PORT", 8080))) + app.run(host="0.0.0.0", port=int(os.environ.get("PORT", 8080))) diff --git a/notifications/renderers.py b/notifications/renderers.py new file mode 100644 index 0000000..1b9c329 --- /dev/null +++ b/notifications/renderers.py @@ -0,0 +1,518 @@ +"""Notification rendering helpers for InteractiveBrokersPlatform.""" + +from __future__ import annotations + +from collections.abc import Mapping +import re + +from notifications.events import RenderedNotification + + +_ZH_REASON_REPLACEMENTS = ( + ("feature snapshot guard blocked execution", "特征快照校验阻止执行"), + ("feature snapshot required", "需要特征快照"), + ("feature snapshot compute failed", "特征快照计算失败"), + ("feature_snapshot_download_failed", "特征快照下载失败"), + ("feature_snapshot_compute_failed", "特征快照计算失败"), + ("feature_snapshot_path_missing", "缺少特征快照路径"), + ("feature_snapshot_missing", "特征快照不存在"), + ("feature_snapshot_stale", "特征快照过旧"), + ("feature_snapshot_manifest_missing", "缺少快照清单"), + ("feature_snapshot_profile_mismatch", "快照策略名不匹配"), + ("feature_snapshot_config_name_mismatch", "快照配置名不匹配"), + ("feature_snapshot_config_path_mismatch", "快照配置路径不匹配"), + ("feature_snapshot_contract_version_mismatch", "快照契约版本不匹配"), + ("soxl_soxx_trend_income", "SOXL/SOXX 半导体趋势收益"), + ("tqqq_growth_income", "TQQQ 增长收益"), + ("global_etf_rotation", "全球 ETF 轮动"), + ("russell_1000_multi_factor_defensive", "罗素1000多因子"), + ("tech_communication_pullback_enhancement", "科技通信回调增强"), + ("qqq_tech_enhancement", "科技通信回调增强"), + ("mega_cap_leader_rotation_aggressive", "Mega Cap 激进龙头轮动"), + ("mega_cap_leader_rotation_dynamic_top20", "Mega Cap 动态 Top20 龙头轮动"), + ("mega_cap_leader_rotation_top50_balanced", "Mega Cap Top50 平衡龙头轮动"), + ("dynamic_mega_leveraged_pullback", "Mega Cap 2x 回调策略"), + ("outside_monthly_execution_window", "当前不在月度执行窗口"), + ("no_execution_window_after_snapshot", "快照后没有可用执行窗口"), + ("no-op", "不执行"), + ("monthly snapshot cadence", "月度快照节奏"), + ("waiting inside execution window", "等待进入执行窗口"), + ("small_account_warning=true", "小账户提示=是"), + ("portfolio_equity=", "净值="), + ("min_recommended_equity=", "建议最低净值="), + ( + "integer_shares_min_position_value_may_prevent_backtest_replication", + "整数股和最小仓位限制可能导致实盘无法完全复现回测", + ), + ( + "integer-share minimum position sizing may prevent backtest replication", + "整数股和最小仓位限制可能导致实盘无法完全复现回测", + ), + ("small account warning: portfolio equity", "小账户提示：净值"), + ("small account warning", "小账户提示"), + ("is below the recommended", "低于建议"), + ("is below recommended", "低于建议"), + ("snapshot_as_of=", "快照日期="), + ("snapshot=", "快照日期="), + ("allowed=", "允许日期="), + ("", "未知"), + ("", "无"), + ("RISK-ON", "风险开启"), + ("DE-LEVER", "降杠杆"), + ("regime=hard_defense", "市场阶段=强防御"), + ("regime=soft_defense", "市场阶段=软防御"), + ("regime=risk_on", "市场阶段=进攻"), + ("benchmark_trend=down", "基准趋势=向下"), + ("benchmark_trend=up", "基准趋势=向上"), + ("benchmark=down", "基准趋势=向下"), + ("benchmark=up", "基准趋势=向上"), + ("breadth=", "市场宽度="), + ("target_stock=", "目标股票仓位="), + ("realized_stock=", "实际股票仓位="), + ("stock_exposure=", "股票目标仓位="), + ("safe_haven=", "避险仓位="), + ("selected=", "入选标的数="), + ("top=", "前排标的="), + ("no_selection", "无入选标的"), + ("outside_execution_window", "当前不在执行窗口"), + ("pending_orders_detected", "检测到未完成订单"), + ("same_day_execution_locked", "当日执行锁已存在"), + ("same_day_fills_detected", "检测到当日成交"), + ("insufficient_buying_power", "购买力不足"), + ("missing_price", "缺少报价"), + ("no_equity", "无净值"), + ("fail_closed", "关闭执行"), + ("reason=", "原因="), + ("fail_reason=", "失败原因="), + ("decision=", "决策="), +) +_DETAIL_FIELD_SPLIT_RE = re.compile(r"\s+(?=[^\s=:：]+[=:：])") + + +def _format_text(value, *, fallback: str) -> str: + text = str(value).strip() if value is not None else "" + return text or fallback + + +def _format_symbol_preview(symbols, *, limit: int = 3) -> str: + normalized = [str(symbol).strip().upper() for symbol in symbols if str(symbol).strip()] + if not normalized: + return "" + shown = normalized[:limit] + remaining = len(normalized) - len(shown) + if remaining > 0: + shown.append(f"+{remaining}") + return ",".join(shown) + + +def _translator_uses_zh(translator) -> bool: + sample = str(translator("no_trades")) + return any("\u4e00" <= ch <= "\u9fff" for ch in sample) + + +def _localize_notification_text(text: str, *, translator) -> str: + value = str(text or "").strip() + if not value or not _translator_uses_zh(translator): + return value + localized = value + for source, target in _ZH_REASON_REPLACEMENTS: + localized = localized.replace(source, target) + return localized + + +def _split_detail_segment(text: str) -> list[str]: + value = str(text or "").strip() + if not value: + return [] + if "=" not in value and ":" not in value and "：" not in value: + return [value] + return [part.strip() for part in _DETAIL_FIELD_SPLIT_RE.split(value) if part.strip()] + + +def _split_labeled_text(text: str) -> list[str]: + segments = [segment.strip() for segment in str(text or "").split(" | ") if segment.strip()] + if not segments: + return [] + lines = [segments[0]] + for segment in segments[1:]: + lines.extend(_split_detail_segment(segment)) + return lines + + +def _format_prefixed_text(prefix: str, text: str) -> list[str]: + parts = _split_labeled_text(text) + if not parts: + return [] + lines = [f"{prefix} {parts[0]}".strip()] + lines.extend(f" - {part}" for part in parts[1:]) + return lines + + +def _summarize_target_changes(target_vs_current, *, limit: int = 5) -> str | None: + rows = [] + for row in target_vs_current or (): + symbol = str(row.get("symbol") or "").strip().upper() + if not symbol: + continue + delta = float(row.get("delta_weight") or 0.0) + if abs(delta) < 0.001: + continue + rows.append((abs(delta), symbol, delta)) + if not rows: + return None + rows.sort(key=lambda item: (-item[0], item[1])) + preview = [f"{symbol} {delta:+.1%}" for _abs_delta, symbol, delta in rows[:limit]] + remaining = len(rows) - len(preview) + if remaining > 0: + preview.append(f"+{remaining}") + return ", ".join(preview) + + +def _summarize_orders(orders, *, limit: int = 3) -> str: + preview = [] + for order in orders[:limit]: + symbol = str(order.get("symbol") or "").strip().upper() + quantity = int(order.get("quantity") or 0) + if symbol and quantity > 0: + preview.append(f"{symbol} {quantity}") + elif symbol: + preview.append(symbol) + remaining = len(orders) - len(preview) + if remaining > 0: + preview.append(f"+{remaining}") + return ", ".join(preview) + + +def _build_order_batch_lines(execution_summary, *, translator) -> list[str]: + mode = str(execution_summary.get("mode") or "").strip().lower() + order_groups = [ + ("orders_submitted", "dry_run" if mode == "dry_run" else "submitted"), + ("orders_filled", "filled"), + ("orders_partially_filled", "partial"), + ] + lines: list[str] = [] + for field_name, prefix in order_groups: + orders = list(execution_summary.get(field_name) or []) + if not orders: + continue + buy_orders = [order for order in orders if str(order.get("side") or "").strip().lower() == "buy"] + sell_orders = [order for order in orders if str(order.get("side") or "").strip().lower() == "sell"] + if buy_orders: + lines.append( + translator( + f"{prefix}_buy_batch", + count=len(buy_orders), + details=_summarize_orders(buy_orders), + ) + ) + if sell_orders: + lines.append( + translator( + f"{prefix}_sell_batch", + count=len(sell_orders), + details=_summarize_orders(sell_orders), + ) + ) + return lines + + +def _build_notification_trade_lines( + trade_logs, + *, + execution_summary, + translator, +) -> list[str]: + lines: list[str] = [] + execution_summary = dict(execution_summary or {}) + + no_op_reason = str(execution_summary.get("no_op_reason") or "").strip() + if no_op_reason.startswith("same_day_execution_locked:"): + lines.append( + translator( + "same_day_execution_locked_notice", + mode=_format_text(execution_summary.get("mode"), fallback=""), + trade_date=_format_text(execution_summary.get("trade_date"), fallback=""), + snapshot_date=_format_text(execution_summary.get("snapshot_as_of"), fallback=""), + ) + ) + + fallback_symbols = tuple(execution_summary.get("snapshot_price_fallback_symbols") or ()) + if execution_summary.get("snapshot_price_fallback_used") and fallback_symbols: + lines.append( + translator( + "dry_run_snapshot_prices", + count=len(fallback_symbols), + symbols=_format_symbol_preview(fallback_symbols), + ) + ) + + target_change_summary = _summarize_target_changes(execution_summary.get("target_vs_current")) + if target_change_summary: + lines.append(translator("target_diff_summary", details=target_change_summary)) + + lines.extend(_build_order_batch_lines(execution_summary, translator=translator)) + + for raw_line in trade_logs or (): + text = _localize_notification_text(str(raw_line).strip(), translator=translator) + if not text: + continue + if text.startswith(("目标差异 ", "target_diff ", "DRY_RUN buy ", "DRY_RUN sell ")): + continue + if text.startswith(("🧪 dry-run估价:", "🧪 dry-run pricing:")): + continue + if "execution_lock_acquired" in text or "已获取执行锁" in text: + continue + if text.startswith(("profile=", "strategy_profile=", "策略=")): + continue + if "same_day_execution_locked" in text or "当日执行锁已存在" in text: + continue + if text not in lines: + lines.extend(_split_labeled_text(text)) + + return lines + + +def _resolve_weight_allocation(signal_metadata, *, required: bool) -> dict: + metadata = dict(signal_metadata or {}) + allocation = dict(metadata.get("allocation") or {}) + if not allocation: + if required: + raise ValueError("IBKR execution requires signal_metadata.allocation") + return {} + if allocation.get("target_mode") != "weight": + raise ValueError("IBKR execution requires allocation.target_mode=weight") + targets = { + str(symbol).strip().upper(): float(weight) + for symbol, weight in dict(allocation.get("targets") or {}).items() + } + return { + "strategy_symbols": tuple(str(symbol) for symbol in allocation.get("strategy_symbols", ())), + "risk_symbols": tuple(str(symbol) for symbol in allocation.get("risk_symbols", ())), + "income_symbols": tuple(str(symbol) for symbol in allocation.get("income_symbols", ())), + "safe_haven_symbols": tuple(str(symbol) for symbol in allocation.get("safe_haven_symbols", ())), + "targets": targets, + } + + +def _format_dashboard_text(text) -> str: + lines = [line.rstrip() for line in str(text or "").strip().splitlines()] + while lines and not lines[0].strip(): + lines.pop(0) + while lines and not lines[-1].strip(): + lines.pop() + return "\n".join(lines) + + +def _strategy_dashboard_text(signal_metadata) -> str: + metadata = signal_metadata if isinstance(signal_metadata, Mapping) else {} + raw_annotations = metadata.get("execution_annotations") + annotations = raw_annotations if isinstance(raw_annotations, Mapping) else {} + return _format_dashboard_text( + annotations.get("dashboard_text") + or metadata.get("dashboard_text") + or metadata.get("dashboard") + or "" + ) + + +def build_dashboard( + positions, + account_values, + signal_desc, + status_desc, + *, + strategy_profile=None, + strategy_display_name=None, + target_weights=None, + signal_metadata=None, + translator, + separator, + status_icon="🐤", +): + signal_metadata = signal_metadata or {} + strategy_dashboard = _strategy_dashboard_text(signal_metadata) + if strategy_dashboard: + return strategy_dashboard + equity = account_values.get("equity", 0) + buying_power = account_values.get("buying_power", 0) + position_lines = [] + for symbol in sorted(positions.keys()): + qty = positions[symbol]["quantity"] + avg = positions[symbol]["avg_cost"] + market_value = qty * avg + position_lines.append(f" - {symbol}: {qty}股 | ${market_value:,.2f}") + position_text = "\n".join(position_lines) if position_lines else translator("empty_positions") + allocation = _resolve_weight_allocation(signal_metadata, required=False) + target_lines = [] + for symbol, weight in sorted(allocation.get("targets", {}).items(), key=lambda item: (-item[1], item[0])): + target_lines.append(f" - {symbol}: {weight:.1%}") + target_text = "\n".join(target_lines) if target_lines else translator("empty_target_weights") + regime = signal_metadata.get("regime") + breadth_ratio = signal_metadata.get("breadth_ratio") + target_stock_weight = signal_metadata.get("target_stock_weight") + realized_stock_weight = signal_metadata.get("realized_stock_weight") + safe_haven_weight = signal_metadata.get("safe_haven_weight") + snapshot_as_of = signal_metadata.get("snapshot_as_of") + strategy_name = _format_text( + strategy_display_name, + fallback=_format_text(strategy_profile, fallback=""), + ) + diagnostics = [ + translator("strategy_label", name=strategy_name), + translator("regime_detail", value=_format_text(regime, fallback="")) if regime is not None else None, + translator("breadth_detail", value=f"{breadth_ratio:.1%}") if isinstance(breadth_ratio, (int, float)) else None, + translator("target_stock_detail", value=f"{target_stock_weight:.1%}") + if isinstance(target_stock_weight, (int, float)) + else None, + translator("realized_stock_detail", value=f"{realized_stock_weight:.1%}") + if isinstance(realized_stock_weight, (int, float)) + and isinstance(target_stock_weight, (int, float)) + and abs(float(realized_stock_weight) - float(target_stock_weight)) >= 0.01 + else None, + translator("safe_haven_target_detail", value=f"{safe_haven_weight:.1%}") + if isinstance(safe_haven_weight, (int, float)) + else None, + translator("snapshot_as_of_detail", value=_format_text(snapshot_as_of, fallback="")) if snapshot_as_of else None, + ] + diagnostics_lines = [f" - {part}" for part in diagnostics if part] + diagnostics_text = "\n".join(diagnostics_lines) + localized_status_desc = _localize_notification_text(status_desc, translator=translator) + localized_signal_desc = _localize_notification_text(signal_desc, translator=translator) + status_lines = _format_prefixed_text(status_icon, localized_status_desc) + signal_lines = _format_prefixed_text("🎯", localized_signal_desc) + status_text = "\n".join(status_lines) + signal_text = "\n".join(signal_lines) + return ( + f"{translator('account_summary_title')}\n" + f" - {translator('equity')}: ${equity:,.2f}\n" + f" - {translator('buying_power')}: ${buying_power:,.2f}\n" + f"{separator}\n" + f"{translator('positions_title')}\n" + f"{position_text}\n" + f"{separator}\n" + f"{translator('execution_summary_title')}\n" + f"{diagnostics_text}\n" + f"{separator}\n" + f"{status_text}\n" + f"{signal_text}\n" + f"{separator}\n" + f"{translator('target_weights_title')}:\n{target_text}" + ) + + +def _first_prefixed_line(prefix: str, text: str, *, translator) -> str | None: + localized = _localize_notification_text(text, translator=translator) + lines = _format_prefixed_text(prefix, localized) + return lines[0] if lines else None + + +def _build_compact_message( + *, + title: str, + strategy_display_name: str | None, + signal_desc: str, + status_desc: str, + status_icon: str, + translator, + separator: str, + body_lines, + dashboard_text: str = "", +) -> str: + lines = [title] + strategy_name = _format_text(strategy_display_name, fallback="") + lines.append(translator("strategy_label", name=strategy_name)) + dashboard = _format_dashboard_text(dashboard_text) + if dashboard: + lines.append(separator) + lines.extend(dashboard.splitlines()) + status_line = _first_prefixed_line(status_icon, status_desc, translator=translator) + if status_line: + lines.append(status_line) + signal_line = _first_prefixed_line("🎯", signal_desc, translator=translator) + if signal_line: + lines.append(signal_line) + compact_body = [str(line).strip() for line in body_lines or () if str(line).strip()] + if compact_body: + lines.append(separator) + lines.extend(compact_body) + return "\n".join(lines) + + +def render_heartbeat_notification( + *, + dashboard, + strategy_dashboard, + no_op_text, + signal_desc, + status_desc, + status_icon, + translator, + separator, + strategy_display_name, +) -> RenderedNotification: + detailed_text = f"{translator('heartbeat_title')}\n{dashboard}\n{separator}\n{no_op_text}" + compact_text = _build_compact_message( + title=translator("heartbeat_title"), + strategy_display_name=strategy_display_name, + signal_desc=signal_desc, + status_desc=status_desc, + status_icon=status_icon, + translator=translator, + separator=separator, + body_lines=[no_op_text], + dashboard_text=strategy_dashboard, + ) + return RenderedNotification(detailed_text=detailed_text, compact_text=compact_text) + + +def render_trade_notification( + *, + dashboard, + strategy_dashboard, + trade_logs, + execution_summary, + signal_desc, + status_desc, + status_icon, + translator, + separator, + strategy_display_name, +) -> RenderedNotification: + if trade_logs: + notification_trade_lines = _build_notification_trade_lines( + trade_logs, + execution_summary=execution_summary, + translator=translator, + ) + detailed_text = ( + f"{translator('rebalance_title')}\n" + f"{dashboard}\n" + f"{separator}\n" + f"{chr(10).join(notification_trade_lines)}" + ) + compact_text = _build_compact_message( + title=translator("rebalance_title"), + strategy_display_name=strategy_display_name, + signal_desc=signal_desc, + status_desc=status_desc, + status_icon=status_icon, + translator=translator, + separator=separator, + body_lines=notification_trade_lines, + dashboard_text=strategy_dashboard, + ) + return RenderedNotification(detailed_text=detailed_text, compact_text=compact_text) + + detailed_text = f"{translator('heartbeat_title')}\n{dashboard}\n{separator}\n{translator('no_trades')}" + compact_text = _build_compact_message( + title=translator("heartbeat_title"), + strategy_display_name=strategy_display_name, + signal_desc=signal_desc, + status_desc=status_desc, + status_icon=status_icon, + translator=translator, + separator=separator, + body_lines=[translator("no_trades")], + dashboard_text=strategy_dashboard, + ) + return RenderedNotification(detailed_text=detailed_text, compact_text=compact_text) diff --git a/tests/test_rebalance_service.py b/tests/test_rebalance_service.py index f2bcbd6..4470518 100644 --- a/tests/test_rebalance_service.py +++ b/tests/test_rebalance_service.py @@ -1,7 +1,7 @@ import json -from application.rebalance_service import build_dashboard, run_strategy_core -from application.rebalance_service import _build_notification_trade_lines +from application.rebalance_service import run_strategy_core +from notifications.renderers import _build_notification_trade_lines, build_dashboard from notifications.telegram import build_translator @@ -226,7 +226,7 @@ def fake_execute_rebalance( strategy_display_name="Global ETF Rotation", ) - assert result == "OK - executed" + assert result.result == "OK - executed" assert observed["strategy_symbols"] == ("AAA", "BOXX") assert observed["signal_metadata"]["managed_symbols"] == ("AAA", "BOXX") assert observed["messages"] @@ -307,7 +307,7 @@ def disconnect(self): reconciliation_output_path=output_path, ) - assert result == "OK - executed" + assert result.result == "OK - executed" payload = json.loads(output_path.read_text(encoding="utf-8")) assert payload["strategy_profile"] == "tech_communication_pullback_enhancement" assert payload["snapshot_as_of"] == "2026-03-31" @@ -355,7 +355,7 @@ def disconnect(self): reconciliation_output_path=output_root, ) - assert result == "OK - heartbeat" + assert result.result == "OK - heartbeat" candidate_paths = [ output_root, output_root / "2026-04-01" / "reconciliation.json", diff --git a/tests/test_request_handling.py b/tests/test_request_handling.py index 9c58d20..feef8f7 100644 --- a/tests/test_request_handling.py +++ b/tests/test_request_handling.py @@ -1,3 +1,6 @@ +from application.cycle_result import StrategyCycleResult + + def test_handle_request_get_returns_safe_message(strategy_module, monkeypatch): def fail_if_called(): raise AssertionError("GET should not execute strategy") @@ -126,8 +129,9 @@ def test_handle_request_enriches_runtime_report_with_cycle_details(strategy_modu monkeypatch.setattr(strategy_module, "is_market_open_today", lambda: True) def fake_run_strategy_core(): - strategy_module.LAST_CYCLE_DETAILS = { - "execution_summary": { + return StrategyCycleResult( + result="OK - executed", + execution_summary={ "execution_status": "executed", "orders_submitted": [{"symbol": "AAA"}], "orders_skipped": [], @@ -136,9 +140,8 @@ def fake_run_strategy_core(): "snapshot_price_fallback_count": 1, "snapshot_price_fallback_symbols": ["AAA"], }, - "reconciliation_record_path": "/tmp/reconciliation.json", - } - return "OK - executed" + reconciliation_record_path="/tmp/reconciliation.json", + ) monkeypatch.setattr(strategy_module, "run_strategy_core", fake_run_strategy_core) monkeypatch.setattr( @@ -233,8 +236,8 @@ def fake_connect_ib(): first = strategy_module.run_strategy_core() second = strategy_module.run_strategy_core() - assert first == "OK - heartbeat" - assert second == "OK - heartbeat" + assert first.result == "OK - heartbeat" + assert second.result == "OK - heartbeat" assert observed["connect_calls"] == 2 assert observed["disconnect_calls"] == 2 diff --git a/tests/test_runtime_composer.py b/tests/test_runtime_composer.py new file mode 100644 index 0000000..dc6c9ad --- /dev/null +++ b/tests/test_runtime_composer.py @@ -0,0 +1,90 @@ +import sys +from pathlib import Path +from types import SimpleNamespace + + +ROOT = Path(__file__).resolve().parents[1] +if str(ROOT) not in sys.path: + sys.path.insert(0, str(ROOT)) + +from application.runtime_composer import IBKRRuntimeComposer # noqa: E402 + + +def test_runtime_composer_builds_runtime_and_config_from_local_builders(): + observed = {} + + def fake_notification_builder(**kwargs): + observed["notification_builder"] = kwargs + return SimpleNamespace(notification_port="notification-port") + + def fake_reporting_builder(**kwargs): + observed["reporting_builder"] = kwargs + return "reporting-adapters" + + composer = IBKRRuntimeComposer( + service_name="interactive-brokers-platform", + strategy_profile="global_etf_rotation", + strategy_domain="us_equity", + account_group="default", + project_id="project-1", + instance_name="ib-gateway", + account_ids=("U123456",), + strategy_target_mode="weight", + strategy_artifact_dir="/tmp/artifacts", + strategy_display_name="Global ETF Rotation", + strategy_display_name_localized="全球 ETF 轮动", + managed_symbols=("AAA", "BIL"), + signal_source="market_data", + status_icon="🐤", + safe_haven="BIL", + dry_run_only=True, + strategy_config_source="env", + ib_gateway_host_resolver=lambda: "127.0.0.1", + ib_gateway_port=4001, + ib_gateway_mode="live", + ib_gateway_ip_mode="internal", + ib_client_id=1, + ib_connect_timeout_seconds=60, + feature_snapshot_path="/tmp/snapshot.csv", + feature_snapshot_manifest_path="/tmp/snapshot.manifest.json", + strategy_config_path="/tmp/config.json", + reconciliation_output_path="/tmp/reconciliation.json", + translator=lambda key, **_kwargs: key, + separator="━━━━━━━━━━━━━━━━━━", + send_message=lambda message: observed.setdefault("sent_message", message), + connect_ib_fn=lambda: "ib-connection", + build_portfolio_snapshot_fn=lambda ib: ("portfolio-snapshot", ib), + compute_signals_fn="compute-signals", + execute_rebalance_fn="execute-rebalance", + run_id_builder=lambda: "run-001", + event_logger="event-logger", + report_builder="report-builder", + report_persister="report-persister", + trace_extractor="trace-extractor", + env_reader=lambda name, default="": { + "EXECUTION_REPORT_OUTPUT_DIR": "/tmp/runtime-reports", + "EXECUTION_REPORT_GCS_URI": "gs://bucket/runtime-reports", + }.get(name, default), + printer=lambda *_args, **_kwargs: None, + notification_builder=fake_notification_builder, + reporting_builder=fake_reporting_builder, + ) + + notification_adapters = composer.build_notification_adapters() + reporting_adapters = composer.build_reporting_adapters() + runtime = composer.build_rebalance_runtime() + config = composer.build_rebalance_config() + + assert notification_adapters.notification_port == "notification-port" + assert observed["notification_builder"]["send_message"] + assert observed["reporting_builder"]["account_scope"] == "default" + assert observed["reporting_builder"]["managed_symbols"] == ("AAA", "BIL") + assert runtime.connect_ib() == "ib-connection" + assert runtime.portfolio_port_factory("ib").get_portfolio_snapshot() == ("portfolio-snapshot", "ib") + assert runtime.compute_signals == "compute-signals" + assert runtime.execute_rebalance == "execute-rebalance" + assert runtime.notifications == "notification-port" + assert config.separator == "━━━━━━━━━━━━━━━━━━" + assert config.strategy_display_name == "全球 ETF 轮动" + assert config.reconciliation_output_path == "/tmp/reconciliation.json" + assert reporting_adapters == "reporting-adapters"