From 4fd5e14fc8eea9f7f53b3a8e73a9a1b860998eae Mon Sep 17 00:00:00 2001 From: Pigbibi <20649888+Pigbibi@users.noreply.github.com> Date: Thu, 10 Sep 2026 06:30:03 +0800 Subject: [PATCH] fix: carry SG next-session signals through durable live execution Co-Authored-By: Codex --- application/durable_execution_commands.py | 173 +++++++++++- application/longbridge_execution.py | 20 ++ application/rebalance_service.py | 310 +++++++++++++++++++++- application/runtime_composer.py | 29 +- application/runtime_dependencies.py | 4 + application/runtime_strategy_adapters.py | 56 ++++ main.py | 4 + tests/test_durable_execution_commands.py | 82 ++++++ tests/test_rebalance_service.py | 93 +++++++ tests/test_request_handling.py | 8 +- tests/test_runtime_strategy_adapters.py | 73 +++++ 11 files changed, 839 insertions(+), 13 deletions(-) diff --git a/application/durable_execution_commands.py b/application/durable_execution_commands.py index c5d6709..696ef94 100644 --- a/application/durable_execution_commands.py +++ b/application/durable_execution_commands.py @@ -1,13 +1,16 @@ -"""Paper-only producer for immutable delayed-execution command evidence.""" +"""Immutable delayed-execution commands with separate paper and live admission.""" from __future__ import annotations import hashlib +import importlib.metadata import json from collections.abc import Mapping +from pathlib import Path from typing import Any from quant_platform_kit.common.execution_commands import ( + EXECUTION_COMMAND_SCHEMA_VERSION, EXECUTION_COMMAND_STRATEGY_RELEASE_FIELD, ExecutionCommand, ExecutionCommandStore, @@ -28,6 +31,7 @@ from quant_platform_kit.common.strategy_release import build_strategy_release_identity PAPER_EXECUTION_INTENT_SCHEMA_VERSION = "longbridge.paper-execution-intent.v1" +LIVE_EXECUTION_COMMAND_ENABLED_ENV = "LONGBRIDGE_DURABLE_EXECUTION_COMMAND_LIVE_ENABLED" def _canonical_json(value: Mapping[str, Any]) -> str: @@ -55,6 +59,165 @@ def _normalized_targets(value: object) -> dict[str, float]: return {symbol: normalized[symbol] for symbol in sorted(normalized)} +def _sha256_text(value: object) -> str: + return hashlib.sha256(str(value or "").encode("utf-8")).hexdigest() + + +def _required_sha256(value: object, *, field_name: str) -> str: + normalized = str(value or "").strip().lower() + if len(normalized) != 64 or any(character not in "0123456789abcdef" for character in normalized): + raise ValueError(f"{field_name} must be a sha256 digest") + return normalized + + +def build_live_runtime_identity_digest(*, strategy_profile: str, runtime_config: Mapping[str, Any]) -> str: + packages = {} + for name in ("quant-platform-kit", "us-equity-strategies"): + distribution = importlib.metadata.distribution(name) + try: + direct_url = json.loads(distribution.read_text("direct_url.json") or "{}") + commit_id = str((direct_url.get("vcs_info") or {}).get("commit_id") or "").lower() + except (TypeError, ValueError, AttributeError): + commit_id = "" + if len(commit_id) != 40 or any(character not in "0123456789abcdef" for character in commit_id): + raise RuntimeError(f"{name} source commit identity is unavailable") + packages[name] = { + "version": distribution.version, + "source_commit": commit_id, + } + identity = { + "strategy_profile": str(strategy_profile or "").strip(), + "runtime_config": dict(runtime_config), + "packages": packages, + } + return hashlib.sha256(_canonical_json(identity).encode("utf-8")).hexdigest() + + +def _build_live_execution_intent( + *, + physical_account_id: str, + runtime_identity_digest: str, + execution: Mapping[str, Any], + allocation: Mapping[str, Any], +) -> dict[str, object]: + account_id = str(physical_account_id or "").strip() + if not account_id: + raise ValueError("physical_account_id is required") + frozen_execution = json.loads(_canonical_json(execution)) + return { + "kind": "longbridge_next_session_live", + "physical_account_digest": _sha256_text(account_id), + "runtime_identity_digest": _required_sha256( + runtime_identity_digest, + field_name="runtime_identity_digest", + ), + "execution": frozen_execution, + "allocation": { + "target_mode": str(allocation.get("target_mode") or "").strip(), + "targets": _normalized_targets(allocation.get("targets")), + "strategy_symbols": _normalized_symbols(allocation.get("strategy_symbols")), + "risk_symbols": _normalized_symbols(allocation.get("risk_symbols")), + "income_symbols": _normalized_symbols(allocation.get("income_symbols")), + "safe_haven_symbols": _normalized_symbols(allocation.get("safe_haven_symbols")), + }, + } + + +def build_live_execution_command( + *, + platform: str, + account_scope: str, + strategy_profile: str, + physical_account_id: str, + runtime_identity_digest: str, + execution: Mapping[str, Any], + allocation: Mapping[str, Any], +) -> ExecutionCommand: + """Build one immutable next-session live intent without broker authority.""" + intent = _build_live_execution_intent( + physical_account_id=physical_account_id, + runtime_identity_digest=runtime_identity_digest, + execution=execution, + allocation=allocation, + ) + return ExecutionCommand.from_decision( + platform=platform, + account_scope=account_scope, + strategy_profile=strategy_profile, + execution_mode="live", + signal_date=execution.get("signal_date"), + effective_date=execution.get("effective_date"), + execution_timing_contract=execution.get("execution_timing_contract"), + decision_digest=hashlib.sha256(_canonical_json(intent).encode("utf-8")).hexdigest(), + intent=intent, + ) + + +def enqueue_live_execution_command( + *, + enabled: bool, + dry_run_only: bool, + store: ExecutionCommandStore | None, + platform: str, + account_scope: str, + strategy_profile: str, + physical_account_id: str, + runtime_identity_digest: str, + execution: Mapping[str, Any], + allocation: Mapping[str, Any], +) -> tuple[ExecutionCommand, bool] | None: + if not enabled: + return None + if dry_run_only: + raise RuntimeError("durable live execution command is live-only") + if store is None or (not store.cloud_prefix_uri and not store.local_dir): + raise RuntimeError("durable live execution command store is required") + command = build_live_execution_command( + platform=platform, + account_scope=account_scope, + strategy_profile=strategy_profile, + physical_account_id=physical_account_id, + runtime_identity_digest=runtime_identity_digest, + execution=execution, + allocation=allocation, + ) + return command, bool(store.enqueue(command)) + + +def list_live_execution_commands(store: ExecutionCommandStore) -> tuple[ExecutionCommand, ...]: + """Read all durable live commands so unresolved prior sessions stay blocking.""" + if store.cloud_prefix_uri: + prefix = "/".join( + ( + str(store.cloud_prefix_uri).rstrip("/"), + store.namespace, + EXECUTION_COMMAND_SCHEMA_VERSION, + ) + ) + object_store = store.object_store or store._object_store() # noqa: SLF001 + locations = tuple(object_store.list(prefix)) + read_text = object_store.read_text + elif store.local_dir: + root = Path(store.local_dir) / store.namespace / EXECUTION_COMMAND_SCHEMA_VERSION + locations = tuple(root.glob("*/*/command.json")) if root.exists() else () + + def read_text(location): + return Path(location).read_text(encoding="utf-8") + else: + raise RuntimeError("execution command store has no durable backend") + commands = ( + ExecutionCommand.from_dict(json.loads(read_text(location))) + for location in locations + if str(location).endswith("/command.json") or Path(str(location)).name == "command.json" + ) + return tuple( + sorted( + (item for item in commands if item.execution_mode == "live"), + key=lambda item: item.command_id, + ) + ) + + def _build_paper_execution_decision_intent( *, allocation: Mapping[str, Any], @@ -217,3 +380,11 @@ def resolve_paper_execution_command_consumer_enabled(*, env_reader, dry_run_only if enabled and not dry_run_only: raise RuntimeError("durable execution command consumer is paper-only and cannot be enabled live") return enabled + + +def resolve_live_execution_command_enabled(*, env_reader, dry_run_only: bool) -> bool: + raw_value = str(env_reader(LIVE_EXECUTION_COMMAND_ENABLED_ENV, "") or "").strip().lower() + enabled = raw_value in {"1", "true", "t", "yes", "y", "on"} + if enabled and dry_run_only: + raise RuntimeError("durable live execution command is live-only") + return enabled diff --git a/application/longbridge_execution.py b/application/longbridge_execution.py index 961048e..9ea897b 100644 --- a/application/longbridge_execution.py +++ b/application/longbridge_execution.py @@ -13,6 +13,26 @@ _qpk_submit_order = None +def fetch_live_order_status(t_ctx: Any, order_id: str) -> dict[str, str] | None: + """Read an exact durable order, including orders from prior sessions.""" + if not str(order_id or "").strip(): + return None + try: + order = t_ctx.order_detail(order_id) + if str(getattr(order, "order_id", "")) != str(order_id): + return None + status = str(getattr(order, "status", "Unknown")).rsplit(".", 1)[-1] + if status == "PartialFilled": + status = "PartiallyFilled" + return { + "status": status, + "executed_qty": str(getattr(order, "executed_quantity", "0")), + "executed_price": str(getattr(order, "executed_price", "0")), + } + except Exception: + return None + + def _get_qpk_submit_order(): global _qpk_submit_order if _qpk_submit_order is None: diff --git a/application/rebalance_service.py b/application/rebalance_service.py index 22dfc4f..9435ccf 100644 --- a/application/rebalance_service.py +++ b/application/rebalance_service.py @@ -2,13 +2,18 @@ from __future__ import annotations +import hashlib import os import re from datetime import datetime from application.execution_service import ExecutionCycleResult, execute_rebalance_cycle from application.execution_state import build_execution_marker_key, claim_account_owner -from application.durable_execution_commands import enqueue_paper_execution_command +from application.durable_execution_commands import ( + enqueue_live_execution_command, + enqueue_paper_execution_command, + list_live_execution_commands, +) from application.paper_strategy_risk_state import record_paper_strategy_risk_state_transition from application.runtime_dependencies import LongBridgeRebalanceConfig, LongBridgeRebalanceRuntime from quant_platform_kit.common.account_identity import ( @@ -17,6 +22,7 @@ evaluate_account_identity, ) from quant_platform_kit.common.models import ExecutionReport +from quant_platform_kit.common.execution_commands import ExecutionCommandState from quant_platform_kit.common.port_adapters import CallableExecutionPort from quant_platform_kit.longbridge.market_data import fetch_lot_sizes from application.signal_snapshot import build_signal_snapshot @@ -88,6 +94,153 @@ def _plan_allocation(plan): def _noop_sleep(_seconds): return None + +def _snapshot_session_date(snapshot) -> str: + value = getattr(snapshot, "as_of", "") + if hasattr(value, "date"): + return value.date().isoformat() + return str(value or "")[:10] + + +def _physical_account_digest(config: LongBridgeRebalanceConfig) -> str: + return hashlib.sha256(_resolve_physical_account_id(config=config).encode("utf-8")).hexdigest() + + +def _matching_live_commands(*, config): + store = getattr(config, "execution_command_store", None) + if store is None: + raise RuntimeError("durable live execution command store is required") + return tuple( + command + for command in list_live_execution_commands(store) + if command.platform == "longbridge" + and command.account_scope == str(config.execution_state_account_scope or "unknown").lower() + and command.strategy_profile == str(config.strategy_profile or "unknown").lower() + and command.execution_mode == "live" + ) + + +def _validate_live_command_binding(*, command, config) -> tuple[dict, dict]: + intent = command.intent + if intent.get("kind") != "longbridge_next_session_live": + raise ValueError("invalid live execution command") + if intent.get("physical_account_digest") != _physical_account_digest(config): + raise ValueError("invalid live execution command") + if intent.get("runtime_identity_digest") != str( + config.durable_execution_runtime_identity_digest or "" + ): + raise ValueError("invalid live execution command") + execution = intent.get("execution") + allocation = intent.get("allocation") + if not isinstance(execution, dict) or not isinstance(allocation, dict): + raise ValueError("invalid live execution command") + if ( + str(execution.get("signal_date") or "") != command.signal_date + or str(execution.get("effective_date") or "") != command.effective_date + or str(execution.get("execution_timing_contract") or "") + != command.execution_timing_contract + or allocation.get("target_mode") != "value" + ): + raise ValueError("invalid live execution command") + return dict(execution), dict(allocation) + + +def _record_live_command_outcome(*, command, store, result) -> None: + pending = tuple(getattr(result, "pending_orders", ()) or ()) + orders = tuple( + { + "broker_order_id": str(item.get("broker_order_id") or ""), + "symbol": str(item.get("symbol") or ""), + "side": str(item.get("side") or ""), + "quantity": float(item.get("quantity") or 0.0), + "submission_status": str(item.get("submission_status") or "").lower(), + } + for item in pending + ) + details = { + "action_done": bool(getattr(result, "action_done", False)), + "orders_count": len(pending), + "orders": orders, + } + if pending and all(str(item.get("submission_status") or "").lower() == "filled" for item in pending): + for state in ( + ExecutionCommandState.SUBMITTED, + ExecutionCommandState.ACCEPTED, + ExecutionCommandState.FILLED, + ): + if store.append_event(command, next_state=state, details=details) is None: + return + return + statuses = {item["submission_status"] for item in orders} + if pending and statuses and statuses <= {"submitted", "accepted"} and all( + item["broker_order_id"] for item in orders + ): + if store.append_event( + command, + next_state=ExecutionCommandState.SUBMITTED, + details=details, + ) is None: + return + if statuses == {"accepted"}: + store.append_event( + command, + next_state=ExecutionCommandState.ACCEPTED, + details=details, + ) + return + next_state = ( + ExecutionCommandState.RECONCILIATION_REQUIRED + if pending or result.action_done + else ExecutionCommandState.REJECTED + if result.execution.get("no_execute") + else ExecutionCommandState.CANCELLED + ) + store.append_event(command, next_state=next_state, details=details) + + +def _reconcile_live_command(*, command, store, trade_context, fetch_order_status) -> None: + state = store.current_state(command) + if state not in { + ExecutionCommandState.SUBMITTED, + ExecutionCommandState.ACCEPTED, + ExecutionCommandState.PARTIALLY_FILLED, + ExecutionCommandState.RECONCILIATION_REQUIRED, + } or not callable(fetch_order_status): + return + events = store.events(command) + orders = tuple((events[-1].details if events else {}).get("orders") or ()) + if not orders or any(not str(item.get("broker_order_id") or "").strip() for item in orders): + return + statuses = [] + for item in orders: + try: + payload = fetch_order_status(trade_context, str(item["broker_order_id"])) + except Exception: + return + if not isinstance(payload, dict): + return + status = str(payload.get("status") or "").strip().lower().replace("_", "") + statuses.append(status) + terminal = {"filled", "cancelled", "canceled", "rejected", "expired"} + if all(status == "filled" for status in statuses): + next_state = ExecutionCommandState.FILLED + elif all(status in terminal for status in statuses): + next_state = ExecutionCommandState.CANCELLED + elif any(status in {"partiallyfilled", "partial"} for status in statuses): + next_state = ExecutionCommandState.PARTIALLY_FILLED + elif all(status in {"accepted", "new", "submitted"} for status in statuses): + next_state = ExecutionCommandState.ACCEPTED + else: + return + if next_state is state: + return + store.append_event( + command, + next_state=next_state, + details={"orders_count": len(orders), "orders": orders}, + expected_previous_state=state, + ) + def _translator_uses_zh(translator) -> bool: return _base_translator_uses_zh(translator) @@ -417,14 +570,115 @@ def load_plan(*, current_snapshot): raise ValueError("LongBridgePlatform requires allocation.target_mode=value") return current_plan, current_portfolio, current_execution, current_allocation + portfolio_port = runtime.portfolio_port_factory(quote_context, trade_context) + + frozen_execution = None + frozen_allocation = None + def fetch_replanned_state(): - current_snapshot = runtime.portfolio_port_factory( - quote_context, - trade_context, - ).get_portfolio_snapshot() + current_snapshot = portfolio_port.get_portfolio_snapshot() + if live_command_claimed and frozen_execution is not None and frozen_allocation is not None: + return_plan = runtime.resolve_frozen_rebalance_plan( + allocation=frozen_allocation, + execution=frozen_execution, + snapshot=current_snapshot, + ) + return ( + return_plan, + _plan_portfolio(return_plan), + _plan_execution(return_plan), + _plan_allocation(return_plan), + ) return load_plan(current_snapshot=current_snapshot) - plan, portfolio, execution, allocation = fetch_replanned_state() + live_command = None + live_command_claimed = False + live_command_blocked = False + live_command_waiting = False + live_command_observation = None + live_command_enabled = bool(getattr(config, "durable_execution_command_live_enabled", False)) + matching_commands = () + if live_command_enabled: + if not getattr(config, "durable_live_execution_session_authorized", False): + raise RuntimeError("durable live execution requires an open exchange session") + matching_commands = _matching_live_commands(config=config) + for command in matching_commands: + _validate_live_command_binding(command=command, config=config) + _reconcile_live_command( + command=command, store=config.execution_command_store, + trade_context=trade_context, fetch_order_status=runtime.fetch_order_status, + ) + # Reconciliation precedes this fresh account read; never size from a snapshot + # captured before an old order was confirmed filled. + initial_snapshot = portfolio_port.get_portfolio_snapshot() + selected_plan = None + if live_command_enabled: + session_date = _snapshot_session_date(initial_snapshot) + today_commands = tuple(c for c in matching_commands if c.signal_date == session_date) + if not today_commands: + # One current signal is saved for the next session independently of + # consuming yesterday's frozen signal. It never routes directly. + selected_plan = load_plan(current_snapshot=initial_snapshot) + _, _, signal_execution, signal_allocation = selected_plan + if str(signal_execution.get("signal_date") or "") != session_date: + raise RuntimeError("live signal session does not match fresh account session") + if not str(signal_execution.get("effective_date") or "") > session_date: + raise RuntimeError("durable live signal must target a future session") + produced = enqueue_live_execution_command( + enabled=True, dry_run_only=config.dry_run_only, + store=config.execution_command_store, platform="longbridge", + account_scope=str(config.execution_state_account_scope or "unknown"), + strategy_profile=str(config.strategy_profile or "unknown"), + physical_account_id=_resolve_physical_account_id(config=config), + runtime_identity_digest=config.durable_execution_runtime_identity_digest, + execution=signal_execution, allocation=signal_allocation, + ) + command, created = produced + matching_commands = (*matching_commands, command) + today_commands = (command,) + live_command_observation = { + "command_id": command.command_id, "status": "QUEUED" if created else "ALREADY_QUEUED", + "effective_date": command.effective_date, + } + else: + live_command_observation = { + "command_id": today_commands[0].command_id, "status": "ALREADY_QUEUED", + "effective_date": today_commands[0].effective_date, + } + terminal = {ExecutionCommandState.FILLED, ExecutionCommandState.CANCELLED, ExecutionCommandState.REJECTED} + states = {c.command_id: config.execution_command_store.current_state(c) for c in matching_commands} + unresolved = tuple(c for c in matching_commands if states[c.command_id] not in terminal) + due = tuple(c for c in unresolved if c.is_due_on(session_date)) + prior_unresolved = tuple(c for c in unresolved if ( + states[c.command_id] is not ExecutionCommandState.QUEUED or c.effective_date < session_date + )) + if prior_unresolved or len(due) > 1 or len(today_commands) > 1: + live_command = (prior_unresolved or due or today_commands)[0] + live_command_blocked = True + elif due: + live_command = due[0] + elif unresolved: + live_command = unresolved[0] + live_command_blocked = True + live_command_waiting = True + if live_command is not None: + frozen_execution, frozen_allocation = _validate_live_command_binding(command=live_command, config=config) + resolver = runtime.resolve_frozen_rebalance_plan + if not callable(resolver): + raise RuntimeError("frozen live decision resolver is required") + frozen_plan = resolver(allocation=frozen_allocation, execution=frozen_execution, snapshot=initial_snapshot) + selected_plan = (frozen_plan, _plan_portfolio(frozen_plan), _plan_execution(frozen_plan), _plan_allocation(frozen_plan)) + if not live_command_blocked: + claim = config.execution_command_store.claim_due( + live_command, as_of_date=session_date, claimant=str(config.strategy_profile), + ) + live_command_claimed = claim is not None + live_command_blocked = not live_command_claimed + if selected_plan is None: + selected_plan = load_plan(current_snapshot=initial_snapshot) + plan, portfolio, execution, allocation = selected_plan + if live_command_observation is not None: + execution["durable_live_execution_command"] = live_command_observation account_identity_blocked = bool( account_identity_decision is not None and not account_identity_decision.broker_write_allowed @@ -471,14 +725,19 @@ def fetch_replanned_state(): execution_marker_key = _build_execution_marker_key(config=config, execution=execution) execution_state_store = getattr(config, "execution_state_store", None) - direct_live_routing_blocked = _direct_live_routing_requires_durable_command( + direct_live_routing_blocked = ( + _direct_live_routing_requires_durable_command( execution=execution, dry_run_only=bool(getattr(config, "dry_run_only", False)), + ) + and not live_command_claimed ) if direct_live_routing_blocked: execution["direct_live_routing_blocked"] = True execution["direct_live_routing_block_reason"] = "durable_execution_command_required" - execution_already_recorded = direct_live_routing_blocked or account_identity_blocked + execution_already_recorded = ( + direct_live_routing_blocked or account_identity_blocked or live_command_blocked + ) dry_run_bypass_marker = _dry_run_bypasses_execution_marker(config) if dry_run_bypass_marker: print( @@ -492,10 +751,17 @@ def fetch_replanned_state(): try: execution_already_recorded = bool(execution_state_store.has_marker(execution_marker_key)) except Exception as exc: + detail = ( + "execution_marker_read_failed" + if live_command_claimed + else f"Marker: {execution_marker_key}\n{type(exc).__name__}: {exc}" + ) runtime.notify_issue( "Execution marker read failed", - f"Marker: {execution_marker_key}\n{type(exc).__name__}: {exc}", + detail, ) + if live_command_claimed: + execution_already_recorded = True if not execution_already_recorded and hasattr(execution_state_store, "has_prior_execution_report"): try: execution_already_recorded = bool( @@ -509,10 +775,17 @@ def fetch_replanned_state(): ) ) except Exception as exc: + detail = ( + "execution_report_dedup_read_failed" + if live_command_claimed + else f"Marker: {execution_marker_key}\n{type(exc).__name__}: {exc}" + ) runtime.notify_issue( "Execution report dedup read failed", - f"Marker: {execution_marker_key}\n{type(exc).__name__}: {exc}", + detail, ) + if live_command_claimed: + execution_already_recorded = True if execution_already_recorded: if account_identity_blocked: @@ -520,6 +793,11 @@ def fetch_replanned_state(): findings=tuple(account_identity_decision.findings), ) runtime.notify_issue("Account identity gate blocked broker orders", message) + elif live_command_waiting: + message = "Durable live execution command queued; waiting for its effective trading session" + elif live_command_blocked: + message = "Durable live execution command is unresolved; broker orders blocked" + runtime.notify_issue("Durable live execution blocked", message) elif direct_live_routing_blocked: message = _durable_command_required_message(execution=execution) runtime.notify_issue("Next-session execution blocked", message) @@ -668,6 +946,18 @@ def submit_claimed_order(order_intent): result=execution_result, notify_issue=runtime.notify_issue, ) + if live_command_claimed and live_command is not None: + try: + _record_live_command_outcome( + command=live_command, + store=config.execution_command_store, + result=execution_result, + ) + except Exception: + runtime.notify_issue( + "Durable live execution outcome write failed", + "durable_live_execution_outcome_persistence_failed", + ) execution = execution_result.execution execution["cash_only_execution"] = bool(getattr(config, "cash_only_execution", True)) execution["signal_snapshot"] = build_signal_snapshot( diff --git a/application/runtime_composer.py b/application/runtime_composer.py index ea45d2b..55462ef 100644 --- a/application/runtime_composer.py +++ b/application/runtime_composer.py @@ -14,7 +14,9 @@ resolve_execution_dedup_enabled, ) from application.durable_execution_commands import ( + build_live_runtime_identity_digest, build_execution_command_store_from_env, + resolve_live_execution_command_enabled, resolve_paper_execution_command_producer_enabled, ) from application.paper_strategy_risk_state import ( @@ -31,6 +33,7 @@ from quant_platform_kit.common.strategy_release import build_runtime_loaded_receipt from notifications.telegram import build_prefixer from quant_platform_kit.notifications.cycle_channel import build_cycle_sender +from application.longbridge_execution import fetch_live_order_status from runtime_execution_policy import FRACTIONAL_BUY_QUANTITY_STEP, dca_compat_mode_enabled, fractional_buy_execution_enabled @@ -147,7 +150,13 @@ def build_notification_adapters(self, *, delivery_events: list[dict[str, Any]] | send_message=self.send_message, notification_channel=self.notification_channel, translator=self.translator, - fetch_order_status=self.fetch_order_status_fn, + fetch_order_status=( + fetch_live_order_status + if resolve_live_execution_command_enabled( + env_reader=self.env_reader, dry_run_only=self.dry_run_only, + ) + else self.fetch_order_status_fn + ), order_poll_interval_sec=self.order_poll_interval_sec, order_poll_max_attempts=self.order_poll_max_attempts, sleeper=self.sleeper, @@ -229,6 +238,11 @@ def build_rebalance_runtime( post_submit_order=notification_adapters.post_submit_order, fetch_order_status=self.fetch_order_status_fn, account_identity_observer=observe_longbridge_account_identity, + resolve_frozen_rebalance_plan=getattr( + self.strategy_adapters, + "resolve_frozen_rebalance_plan", + None, + ), ) def build_read_only_broker_contexts(self) -> tuple[Any, Any]: @@ -255,6 +269,7 @@ def build_rebalance_config( strategy_plugin_error: str | None = None, notification_title_key: str = "", cash_only_execution: bool = True, + live_execution_session_authorized: bool = False, ) -> LongBridgeRebalanceConfig: market_scope_line = self.translator( "market_scope_detail", @@ -284,6 +299,12 @@ def build_rebalance_config( raise RuntimeError( "LongBridge live execution requires a gs:// execution state URI for atomic claims" ) + live_command_enabled = resolve_live_execution_command_enabled( + env_reader=self.env_reader, + dry_run_only=self.dry_run_only, + ) + if live_command_enabled and self.strategy_profile != "soxl_soxx_trend_income": + raise RuntimeError("durable live execution command is only verified for the SOXL profile") return LongBridgeRebalanceConfig( limit_sell_discount=self.limit_sell_discount, limit_buy_premium=self.limit_buy_premium, @@ -322,10 +343,16 @@ def build_rebalance_config( env_reader=self.env_reader, dry_run_only=self.dry_run_only, ), + durable_execution_command_live_enabled=live_command_enabled, + durable_live_execution_session_authorized=bool(live_execution_session_authorized), execution_command_store=build_execution_command_store_from_env( env_reader=self.env_reader, gcp_project_id=self.project_id, ), + durable_execution_runtime_identity_digest=build_live_runtime_identity_digest( + strategy_profile=self.strategy_profile, + runtime_config=getattr(self.strategy_adapters, "strategy_runtime_config", {}), + ) if live_command_enabled else "", strategy_risk_state_paper_enabled=resolve_paper_strategy_risk_state_enabled( env_reader=self.env_reader, dry_run_only=self.dry_run_only, diff --git a/application/runtime_dependencies.py b/application/runtime_dependencies.py index b4b8e74..51126e3 100644 --- a/application/runtime_dependencies.py +++ b/application/runtime_dependencies.py @@ -39,7 +39,10 @@ class LongBridgeRebalanceConfig: execution_state_account_scope: str = "" physical_account_id: str = "" durable_execution_command_paper_enabled: bool = False + durable_execution_command_live_enabled: bool = False + durable_live_execution_session_authorized: bool = False execution_command_store: Any = None + durable_execution_runtime_identity_digest: str = "" strategy_risk_state_paper_enabled: bool = False strategy_risk_state_store: Any = None runtime_release_receipt: Mapping[str, Any] | None = None @@ -62,3 +65,4 @@ class LongBridgeRebalanceRuntime: post_submit_order: Callable[[Any, Any, Any], None] | None = None fetch_order_status: Callable[..., Any] | None = None account_identity_observer: Callable[[Any], Any] | None = None + resolve_frozen_rebalance_plan: Callable[..., dict[str, Any]] | None = None diff --git a/application/runtime_strategy_adapters.py b/application/runtime_strategy_adapters.py index 938d08c..5871297 100644 --- a/application/runtime_strategy_adapters.py +++ b/application/runtime_strategy_adapters.py @@ -13,6 +13,9 @@ should_alert_strategy_plugin_signal, translate_strategy_plugin_value, ) +from quant_platform_kit.common.strategy_contracts import PositionTarget, StrategyDecision +from quant_platform_kit.risk.gate import apply_risk_gate, enrich_decision_risk_diagnostics +from quant_platform_kit.risk.portfolio_diagnostics import extract_portfolio_risk_diagnostics def _get_direct_market_history_profiles() -> frozenset[str]: @@ -198,6 +201,59 @@ def resolve_rebalance_plan(self, *, indicators, snapshot=None, account_state=Non runtime_metadata=runtime_metadata, ) + def resolve_frozen_rebalance_plan(self, *, allocation, execution, snapshot): + """Re-map one stored target decision against a fresh broker snapshot.""" + snapshot = self.strategy_runtime._stamp_portfolio_risk_metadata( # noqa: SLF001 + {"portfolio_snapshot": snapshot} + )["portfolio_snapshot"] + targets = dict(allocation.get("targets") or {}) + roles = {} + for field, role in ( + ("risk_symbols", "risk"), + ("income_symbols", "income"), + ("safe_haven_symbols", "safe_haven"), + ): + for symbol in allocation.get(field, ()) or (): + roles[str(symbol).strip().upper()] = role + decision = StrategyDecision( + positions=tuple( + PositionTarget( + symbol=str(symbol).strip().upper(), + target_value=float(target), + role=roles.get(str(symbol).strip().upper()), + ) + for symbol, target in sorted(targets.items()) + ), + diagnostics={"execution_annotations": dict(execution)}, + ) + # Preserve the same portfolio diagnostics enrichment as the UES gate. + diagnostics = extract_portfolio_risk_diagnostics(snapshot) + decision = enrich_decision_risk_diagnostics( + decision, + unrealized_pnl_pct=diagnostics.get("unrealized_pnl_pct"), + consecutive_losses=diagnostics.get("consecutive_losses"), + ) + capabilities = self.strategy_runtime._build_capital_base_capabilities( # noqa: SLF001 + {"portfolio_snapshot": snapshot} + ) + decision = apply_risk_gate( + decision, + portfolio_snapshot=snapshot, + max_single_weight=0.20, + enforce_value_target_exposure=True, + **capabilities, + ) + runtime_metadata = {"execution_annotations": dict(execution)} + if self.execution_policy is not None: + runtime_metadata["longbridge_execution_policy"] = dict(self.execution_policy) + return self.map_strategy_decision_to_plan_fn( + decision, + snapshot=snapshot, + account_state=None, + strategy_profile=self.strategy_profile, + runtime_metadata=runtime_metadata, + ) + def build_runtime_strategy_adapters( *, diff --git a/main.py b/main.py index 6dc7da7..1f6afee 100644 --- a/main.py +++ b/main.py @@ -353,6 +353,9 @@ def _summarize_cycle_result_for_report(cycle_result, *, dry_run: bool) -> dict: durable_command = execution.get("durable_execution_command") if isinstance(durable_command, dict): summary["durable_execution_command"] = dict(durable_command) + live_command = execution.get("durable_live_execution_command") + if isinstance(live_command, dict): + summary["durable_live_execution_command"] = dict(live_command) account_identity = execution.get("account_identity") if isinstance(account_identity, dict): summary["account_identity"] = dict(account_identity) @@ -789,6 +792,7 @@ def run_strategy(*, force_run: bool = False, validation_only: bool = False, vali else "" ), cash_only_execution=CASH_ONLY_EXECUTION, + live_execution_session_authorized=bool(market_open and not validation_only), ) failure_phase = "strategy_cycle" cycle_result = run_rebalance_cycle(runtime=rebalance_runtime, config=rebalance_config) diff --git a/tests/test_durable_execution_commands.py b/tests/test_durable_execution_commands.py index 05e2c17..45921cb 100644 --- a/tests/test_durable_execution_commands.py +++ b/tests/test_durable_execution_commands.py @@ -9,11 +9,14 @@ sys.path.insert(0, str(ROOT)) from application.durable_execution_commands import ( # noqa: E402 + build_live_execution_command, build_paper_execution_decision_digest, build_paper_execution_command, enqueue_paper_execution_command, + enqueue_live_execution_command, resolve_paper_execution_command_consumer_enabled, resolve_paper_execution_command_producer_enabled, + resolve_live_execution_command_enabled, ) from quant_platform_kit.common.paper_execution_admission import build_paper_risk_admission_receipt from quant_platform_kit.common.strategy_release import build_runtime_loaded_receipt @@ -245,3 +248,82 @@ def test_paper_consumer_rejects_live_enablement() -> None: assert "paper-only" in str(exc) else: # pragma: no cover raise AssertionError("live enablement must fail closed") + + +def test_live_command_is_content_addressed_and_binds_runtime_and_account() -> None: + command = build_live_execution_command( + platform="longbridge", + account_scope="SG", + strategy_profile="soxl_soxx_trend_income", + physical_account_id="account-123", + runtime_identity_digest="a" * 64, + execution={**_execution(), "trade_threshold_value": 100.0}, + allocation=_allocation(), + ) + repeated = build_live_execution_command( + platform="longbridge", + account_scope="SG", + strategy_profile="soxl_soxx_trend_income", + physical_account_id="account-123", + runtime_identity_digest="a" * 64, + execution={**_execution(), "trade_threshold_value": 100.0}, + allocation=_allocation(), + ) + + assert command.command_id == repeated.command_id + assert command.execution_mode == "live" + assert command.intent["runtime_identity_digest"] == "a" * 64 + assert command.intent["physical_account_digest"] != "account-123" + assert "account-123" not in command.intent_json + assert command.intent["allocation"]["targets"] == {"BOXX": 150.0, "SOXL": 350.0} + + +def test_live_flag_is_default_off_and_rejects_paper_runtime() -> None: + assert not resolve_live_execution_command_enabled( + env_reader=lambda _name, default="": default, + dry_run_only=False, + ) + assert resolve_live_execution_command_enabled( + env_reader=lambda _name, _default="": "true", + dry_run_only=False, + ) + try: + resolve_live_execution_command_enabled( + env_reader=lambda _name, _default="": "true", + dry_run_only=True, + ) + except RuntimeError as exc: + assert "live-only" in str(exc) + else: # pragma: no cover + raise AssertionError("paper runtime must fail closed") + + +def test_live_producer_enqueue_is_business_idempotent() -> None: + observed = [] + + class Store: + cloud_prefix_uri = "gs://live/commands" + local_dir = None + + def enqueue(self, command): + observed.append(command.command_id) + return len(observed) == 1 + + kwargs = dict( + enabled=True, + dry_run_only=False, + store=Store(), + platform="longbridge", + account_scope="SG", + strategy_profile="soxl_soxx_trend_income", + physical_account_id="account-123", + runtime_identity_digest="a" * 64, + execution=_execution(), + allocation=_allocation(), + ) + first = enqueue_live_execution_command(**kwargs) + second = enqueue_live_execution_command(**kwargs) + + assert first and first[1] is True + assert second and second[1] is False + assert observed[0] == observed[1] diff --git a/tests/test_rebalance_service.py b/tests/test_rebalance_service.py index 8aa6c95..3323aea 100644 --- a/tests/test_rebalance_service.py +++ b/tests/test_rebalance_service.py @@ -34,6 +34,7 @@ from application.longbridge_portfolio import fetch_strategy_account_state from application.runtime_dependencies import LongBridgeRebalanceConfig, LongBridgeRebalanceRuntime from quant_platform_kit.common.account_identity import BrokerAccountIdentity + from quant_platform_kit.common.execution_commands import ExecutionCommandState, ExecutionCommandStore from notifications.telegram import build_translator from quant_platform_kit.common.models import ExecutionReport, PortfolioSnapshot, Position, QuoteSnapshot from quant_platform_kit.common.port_adapters import CallableExecutionPort, CallableMarketDataPort, CallableNotificationPort, CallablePortfolioPort @@ -1507,6 +1508,98 @@ def test_run_strategy_blocks_live_next_session_decision_without_routing(self): self.assertIn("signal_date=2026-07-17", alerts[0][1]) self.assertIn("effective_date=2026-07-20", alerts[0][1]) + def test_live_commands_continue_daily_and_block_old_unknown_orders(self): + plan = _build_plan( + strategy_symbols=("SOXL",), risk_symbols=("SOXL",), + targets={"SOXL": 400.0}, market_values={"SOXL": 0.0}, + sellable_quantities={"SOXL": 0}, quantities={"SOXL": 0}, + current_min_trade=10.0, trade_threshold_value=10.0, + investable_cash=500.0, available_cash=500.0, total_strategy_equity=500.0, + market_status="Risk on", deploy_ratio_text="70.0%", income_ratio_text="0.0%", + income_locked_ratio_text="0.0%", signal_message="SOXL target", + portfolio_rows=(("SOXL",),), signal_date="2026-07-17", effective_date="2026-07-20", + ) + command_store = ExecutionCommandStore(local_dir=self.enterContext(TemporaryDirectory())) + marker_store = ExecutionMarkerStore(local_dir=self.enterContext(TemporaryDirectory())) + clock = {"day": "2026-07-17", "status": "New"} + next_days = {"2026-07-17": "2026-07-20", "2026-07-20": "2026-07-21", "2026-07-21": "2026-07-22"} + targets = {"2026-07-17": 400.0, "2026-07-20": 300.0, "2026-07-21": 200.0} + resolved_new_signals, orders, alerts, frozen_targets = [], [], [], [] + + def new_plan(**_kwargs): + resolved_new_signals.append(clock["day"]) + return {**plan, "allocation": {**plan["allocation"], "targets": {"SOXL": targets[clock["day"]]}}, + "execution": {**plan["execution"], "signal_date": clock["day"], "effective_date": next_days[clock["day"]]}} + + def frozen_plan(*, allocation, execution, snapshot): + frozen_targets.append(allocation["targets"]["SOXL"]) + return {**plan, "allocation": dict(allocation), "execution": dict(execution)} + + runtime = LongBridgeRebalanceRuntime( + bootstrap=lambda: ("quote", "trade", {"trend": "ok"}), + resolve_rebalance_plan=new_plan, resolve_frozen_rebalance_plan=frozen_plan, + market_data_port_factory=lambda _context: CallableMarketDataPort( + quote_loader=lambda symbol: QuoteSnapshot(symbol=symbol, as_of=clock["day"], last_price=100.0)), + estimate_max_purchase_quantity=lambda *_args, **_kwargs: 5, + notifications=CallableNotificationPort(lambda _message: None), + notify_issue=lambda title, detail: alerts.append((title, detail)), + portfolio_port_factory=lambda *_contexts: CallablePortfolioPort( + lambda: replace(_build_snapshot(plan), as_of=clock["day"])), + execution_port_factory=lambda _context: CallableExecutionPort( + lambda intent: (orders.append(intent), ExecutionReport( + symbol=intent.symbol, side=intent.side, quantity=intent.quantity, + status="accepted", broker_order_id=f"broker-{len(orders)}"))[1]), + fetch_order_status=lambda _context, _order_id: {"status": clock["status"]}, + ) + config = LongBridgeRebalanceConfig( + limit_sell_discount=0.995, limit_buy_premium=1.005, separator="-", + translator=build_translator("en"), with_prefix=lambda message: message, + strategy_profile="soxl_soxx_trend_income", execution_state_account_scope="SG", + physical_account_id="lb-sg-001", dry_run_only=False, notify_no_trade_cycles=False, + execution_dedup_enabled=True, execution_state_store=marker_store, + durable_execution_command_live_enabled=True, execution_command_store=command_store, + durable_live_execution_session_authorized=True, + durable_execution_runtime_identity_digest="a" * 64, + ) + first = rebalance_service.run_strategy(runtime=runtime, config=config) + rebalance_service.run_strategy(runtime=runtime, config=config) + self.assertFalse(first.action_done) + self.assertEqual(first.execution["durable_live_execution_command"]["status"], "QUEUED") + self.assertEqual(alerts, []) + self.assertEqual(resolved_new_signals, ["2026-07-17"]) + monday = command_store.list_due("2026-07-20")[0] + clock["day"] = "2026-07-20" + consumed = rebalance_service.run_strategy(runtime=runtime, config=config) + self.assertTrue(consumed.action_done) + self.assertEqual(consumed.allocation["targets"]["SOXL"], 400.0) + self.assertEqual(len(command_store.list_due("2026-07-21")), 1) + self.assertEqual(len(orders), 1) + clock["day"] = "2026-07-21" + blocked = rebalance_service.run_strategy(runtime=runtime, config=config) + self.assertFalse(blocked.action_done) + self.assertEqual(len(orders), 1) # Yesterday's unresolved order blocks today's due command. + clock["status"] = "Filled" + resumed = rebalance_service.run_strategy(runtime=runtime, config=config) + self.assertTrue(resumed.action_done) + self.assertEqual(resumed.allocation["targets"]["SOXL"], 300.0) + self.assertEqual(len(orders), 2) + self.assertIs(command_store.current_state(monday), ExecutionCommandState.FILLED) + self.assertEqual(resolved_new_signals, list(next_days)) + self.assertEqual(len(command_store.list_due("2026-07-22")), 1) + + def test_live_order_detail_normalizes_real_sdk_enum_and_checks_identity(self): + from application.longbridge_execution import fetch_live_order_status + from longport.openapi import OrderStatus + context = types.SimpleNamespace(order_detail=Mock(return_value=types.SimpleNamespace( + order_id="broker-1", status=OrderStatus.Filled, + executed_quantity="3", executed_price="100", msg=""))) + self.assertEqual(fetch_live_order_status(context, "broker-1")["status"], "Filled") + context.order_detail.assert_called_once_with("broker-1") + context.order_detail.return_value.order_id = "different-order" + self.assertIsNone(fetch_live_order_status(context, "broker-1")) + context.order_detail.side_effect = RuntimeError("private broker failure") + self.assertIsNone(fetch_live_order_status(context, "broker-1")) + def test_run_strategy_skips_when_execution_marker_already_exists(self): sent_messages = [] checked_keys = [] diff --git a/tests/test_request_handling.py b/tests/test_request_handling.py index f4ece8e..5c40ab2 100644 --- a/tests/test_request_handling.py +++ b/tests/test_request_handling.py @@ -936,6 +936,7 @@ def build_rebalance_config( strategy_plugin_error=None, notification_title_key="", cash_only_execution=True, + live_execution_session_authorized=False, ): return types.SimpleNamespace() @@ -1010,6 +1011,7 @@ def build_rebalance_config( strategy_plugin_error=None, notification_title_key="", cash_only_execution=True, + live_execution_session_authorized=False, ): observed["notification_title_key"] = notification_title_key return types.SimpleNamespace() @@ -1060,6 +1062,7 @@ def build_rebalance_config( strategy_plugin_error=None, notification_title_key="", cash_only_execution=True, + live_execution_session_authorized=False, ): observed["notification_title_key"] = notification_title_key return types.SimpleNamespace() @@ -1152,7 +1155,9 @@ def test_cycle_result_summary_keeps_broker_submission_pending_until_reconciled(s skip_logs=(), note_logs=(), action_done=True, - execution={}, + execution={"durable_live_execution_command": { + "command_id": "next-session-command", "status": "QUEUED", + }}, dry_run_orders=(), pending_orders=( { @@ -1174,6 +1179,7 @@ def test_cycle_result_summary_keeps_broker_submission_pending_until_reconciled(s self.assertEqual(summary["order_events_count"], 0) self.assertEqual(summary["orders_pending_count"], 1) self.assertEqual(summary["orders_pending"][0]["broker_order_id"], "lb-order-pending") + self.assertEqual(summary["durable_live_execution_command"]["status"], "QUEUED") def test_notification_delivery_log_summary_records_sent_dry_run_without_raw_text(self): module = load_module() diff --git a/tests/test_runtime_strategy_adapters.py b/tests/test_runtime_strategy_adapters.py index 815c755..686d61b 100644 --- a/tests/test_runtime_strategy_adapters.py +++ b/tests/test_runtime_strategy_adapters.py @@ -3,6 +3,7 @@ from types import SimpleNamespace import pandas as pd +from unittest.mock import patch ROOT = Path(__file__).resolve().parents[1] @@ -249,6 +250,78 @@ def fake_map_plan(decision, **kwargs): } +def test_frozen_plan_reapplies_risk_gate_and_maps_only_the_original_target() -> None: + observed = {} + stamped = SimpleNamespace(total_equity=1000, positions=(), metadata={ + "unrealized_pnl_pct": -0.25, "consecutive_losses": 6, + }) + runtime = SimpleNamespace( + _stamp_portfolio_risk_metadata=lambda inputs: ( + observed.setdefault("stamp_inputs", inputs), {"portfolio_snapshot": stamped}, + )[1], + _build_capital_base_capabilities=lambda inputs: ( + observed.setdefault("capital_inputs", inputs), + {"capital_base": "capital", "capital_base_binding": "binding"}, + )[1] + ) + + def map_plan(decision, **kwargs): + observed["decision"] = decision + observed["map_kwargs"] = kwargs + return {"mapped": True} + + adapters = build_runtime_strategy_adapters( + strategy_runtime=runtime, + strategy_profile="soxl_soxx_trend_income", + strategy_runtime_config={"fixed_param": 0.65}, + available_inputs=("portfolio_snapshot",), + benchmark_symbol="SOXX", + signal_text_fn=str, + translator=str, + broker_adapters=SimpleNamespace(), + calculate_rotation_indicators_fn=lambda *_args, **_kwargs: {}, + build_strategy_evaluation_inputs_fn=lambda **kwargs: kwargs, + map_strategy_decision_to_plan_fn=map_plan, + ) + snapshot = object() + + with patch( + "application.runtime_strategy_adapters.apply_risk_gate", + side_effect=lambda decision, **kwargs: (observed.setdefault("risk_kwargs", kwargs), decision)[1], + ): + result = adapters.resolve_frozen_rebalance_plan( + allocation={ + "targets": {"SOXL": 200.0, "BOXX": 800.0}, + "risk_symbols": ["SOXL"], + "safe_haven_symbols": ["BOXX"], + }, + execution={ + "signal_date": "2026-09-09", + "effective_date": "2026-09-10", + "execution_timing_contract": "next_trading_day", + }, + snapshot=snapshot, + ) + + assert result == {"mapped": True} + assert observed["stamp_inputs"] == {"portfolio_snapshot": snapshot} + assert observed["capital_inputs"] == {"portfolio_snapshot": stamped} + assert observed["decision"].diagnostics["unrealized_pnl_pct"] == -0.25 + assert observed["decision"].diagnostics["consecutive_losses"] == 6 + assert observed["risk_kwargs"] == { + "portfolio_snapshot": stamped, + "max_single_weight": 0.20, + "enforce_value_target_exposure": True, + "capital_base": "capital", + "capital_base_binding": "binding", + } + assert [(item.symbol, item.target_value) for item in observed["decision"].positions] == [ + ("BOXX", 800.0), + ("SOXL", 200.0), + ] + assert observed["map_kwargs"]["runtime_metadata"]["execution_annotations"]["signal_date"] == "2026-09-09" + + def test_runtime_strategy_adapters_loads_and_reports_plugin_signals(): observed = {} signal = SimpleNamespace(