fix: consume completed-session SOXL live commands same day

application/rebalance_service.py

@@ -111,6 +111,30 @@ def _snapshot_session_date(snapshot) -> str:
     return str(value or "")[:10]
 
 
+def _is_completed_session_soxl_command(command) -> bool:
+    execution = command.intent.get("execution") if isinstance(command.intent, dict) else None
+    if (
+        not isinstance(execution, dict)
+        or str(execution.get("completed_session_date") or "") != command.signal_date
+    ):
+        return False
+    try:
+        from application.runtime_strategy_adapters import _next_xnys_session_date
+
+        return _next_xnys_session_date(command.signal_date) == command.effective_date
+    except (TypeError, ValueError):
+        return False
+
+
+def _is_completed_session_soxl_plan(*, execution: dict, session_date: str, strategy_profile: str) -> bool:
+    return (
+        strategy_profile == "soxl_soxx_trend_income"
+        and str(execution.get("completed_session_date") or "") == str(execution.get("signal_date") or "")
+        and str(execution.get("signal_date") or "") < session_date
+        and str(execution.get("effective_date") or "") == session_date
+    )
+
+
 def _physical_account_digest(config: LongBridgeRebalanceConfig) -> str:
     return hashlib.sha256(_resolve_physical_account_id(config=config).encode("utf-8")).hexdigest()
 
@@ -665,47 +689,106 @@ def fetch_replanned_state():
     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,
-                strategy_release=getattr(config, "expected_strategy_release", None),
+        terminal = {ExecutionCommandState.FILLED, ExecutionCommandState.CANCELLED, ExecutionCommandState.REJECTED}
+        states = {c.command_id: config.execution_command_store.current_state(c) for c in matching_commands}
+        unresolved_before_enqueue = tuple(c for c in matching_commands if states[c.command_id] not in terminal)
+        completed_session_path = str(config.strategy_profile or "") == "soxl_soxx_trend_income"
+        if completed_session_path:
+            # A same-day consumer is safe only for commands minted with the
+            # completed-session marker. Old future-dated commands remain
+            # immutable and are blocked rather than reinterpreted as fresh.
+            current_session_commands = tuple(
+                c for c in matching_commands if c.effective_date == session_date
+            )
+            terminal_today = tuple(c for c in current_session_commands if states[c.command_id] in terminal)
+            unsafe_due = tuple(
+                c for c in unresolved_before_enqueue
+                if c.effective_date == session_date and not _is_completed_session_soxl_command(c)
             )
-            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,
-            }
+            today_commands = tuple(
+                c for c in unresolved_before_enqueue
+                if c.effective_date == session_date and _is_completed_session_soxl_command(c)
+            )
+            if terminal_today:
+                # A command already ended in this session remains its own
+                # immutable record. Do not evaluate a changed target or mint a
+                # replacement decision for the same session.
+                live_command = terminal_today[0]
+                live_command_blocked = True
+            elif unsafe_due:
+                live_command = unsafe_due[0]
+                live_command_blocked = True
+            elif not today_commands:
+                selected_plan = load_plan(current_snapshot=initial_snapshot)
+                _, _, signal_execution, signal_allocation = selected_plan
+                if not _is_completed_session_soxl_plan(
+                    execution=signal_execution,
+                    session_date=session_date,
+                    strategy_profile=str(config.strategy_profile or ""),
+                ):
+                    raise RuntimeError("production SOXL live signal requires a completed session proof")
+                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,
+                    strategy_release=getattr(config, "expected_strategy_release", None),
+                )
+                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,
+                }
         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}
+            # Existing non-SOXL and V7 queues retain their next-session flow.
+            today_commands = tuple(c for c in matching_commands if c.signal_date == session_date)
+            if not today_commands:
+                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,
+                    strategy_release=getattr(config, "expected_strategy_release", None),
+                )
+                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,
+                }
         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]
+        if live_command_blocked or prior_unresolved or len(due) > 1 or len(today_commands) > 1:
+            live_command = live_command or (prior_unresolved or due or today_commands)[0]
             live_command_blocked = True
         elif due:
             live_command = due[0]

application/runtime_strategy_adapters.py

@@ -4,6 +4,7 @@
 
 from collections.abc import Collection, Mapping, Callable
 from dataclasses import dataclass, replace
+from datetime import datetime, timedelta, timezone
 from typing import Any
 
 from quant_platform_kit.common.strategy_plugins import (
@@ -26,6 +27,31 @@
 _V7_PROFILE_NAME = "soxl_soxx_core_only_p2_v7_longterm_compounding_cash_reserve"
 
 
+def _completed_xnys_session_date(now: datetime | None = None) -> str:
+    """Resolve the latest completed XNYS session, including early closes."""
+    import pandas as pd
+    import pandas_market_calendars as mcal
+
+    instant = now or datetime.now(timezone.utc)
+    instant = pd.Timestamp(instant)
+    instant = instant.tz_localize("UTC") if instant.tzinfo is None else instant.tz_convert("UTC")
+    schedule = mcal.get_calendar("NYSE").schedule(
+        start_date=(instant - timedelta(days=10)).date(), end_date=instant.date()
+    )
+    completed = schedule[schedule["market_close"] <= instant]
+    if completed.empty:
+        raise RuntimeError("no completed XNYS session is available")
+    return str(completed.index[-1].date())
+
+
+def _next_xnys_session_date(session_date: str) -> str:
+    import pandas_market_calendars as mcal
+
+    schedule = mcal.get_calendar("NYSE").schedule(start_date=session_date, end_date=(datetime.fromisoformat(session_date) + timedelta(days=10)).date())
+    sessions = [str(index.date()) for index in schedule.index]
+    return next(date for date in sessions if date > session_date)
+
+
 def _typed_execution_candidate(materials: Mapping[str, Any]) -> CandidateRiskIdentity | None:
     candidate = materials.get("candidate_risk_identity")
     if isinstance(candidate, CandidateRiskIdentity):
@@ -166,6 +192,12 @@ def calculate_strategy_indicators(self, quote_context):
         if "benchmark_history" in available_inputs or "qqq_history" in available_inputs:
             return self.broker_adapters.build_price_history(market_data_port, self.benchmark_symbol)
         trend_ma_window = int(self.strategy_runtime_config.get("trend_ma_window", 150))
+        if self.strategy_profile == "soxl_soxx_trend_income":
+            return self.calculate_rotation_indicators_fn(
+                quote_context,
+                trend_window=trend_ma_window,
+                completed_session_date=_completed_xnys_session_date(),
+            )
         return self.calculate_rotation_indicators_fn(quote_context, trend_window=trend_ma_window)
 
     def _market_history_symbols(self) -> tuple[str, ...]:
@@ -267,15 +299,38 @@ def resolve_rebalance_plan(self, *, indicators, snapshot=None, account_state=Non
             snapshot=resolved_snapshot,
         )
         runtime_metadata = dict(getattr(evaluation, "metadata", None) or {})
+        completed = indicators.get("completed_session") if isinstance(indicators, Mapping) else None
+        if (
+            self.strategy_profile == "soxl_soxx_trend_income"
+            and isinstance(completed, Mapping)
+            and completed.get("date")
+        ):
+            signal_date = str(completed["date"])
+            annotations = dict(runtime_metadata.get("execution_annotations") or {})
+            annotations.update({"signal_date": signal_date, "effective_date": _next_xnys_session_date(signal_date)})
+            runtime_metadata["execution_annotations"] = annotations
         if self.execution_policy is not None:
             runtime_metadata["longbridge_execution_policy"] = dict(self.execution_policy)
-        return self.map_strategy_decision_to_plan_fn(
+        plan = self.map_strategy_decision_to_plan_fn(
             decision,
             account_state=resolved_account_state if "account_state" in available_inputs else None,
             snapshot=resolved_snapshot,
             strategy_profile=self.strategy_profile,
             runtime_metadata=runtime_metadata,
         )
+        # This is deliberately an opt-in production SOXL path. The mapper may
+        # retain strategy diagnostics over runtime annotations, so stamp the
+        # final mapped execution payload that becomes the immutable command.
+        if self.strategy_profile == "soxl_soxx_trend_income" and isinstance(completed, Mapping) and completed.get("date"):
+            execution = plan.get("execution") if isinstance(plan, Mapping) else None
+            if isinstance(execution, dict):
+                execution.update({
+                    "signal_date": signal_date,
+                    "effective_date": _next_xnys_session_date(signal_date),
+                    "completed_session_date": signal_date,
+                    "generated_at": datetime.now(timezone.utc).isoformat(),
+                })
+        return plan
 
     def resolve_frozen_rebalance_plan(self, *, allocation, execution, snapshot):
         """Re-map one stored target decision against a fresh broker snapshot."""

pyproject.toml

@@ -18,7 +18,7 @@ dependencies = [
     "google-cloud-storage",
     "google-auth",
     "longport==3.0.23",
-    "quant-platform-kit @ git+https://github.com/QuantStrategyLab/QuantPlatformKit.git@d51bb79dea8dc5773384d21b06e726620276558c",
+    "quant-platform-kit @ git+https://github.com/QuantStrategyLab/QuantPlatformKit.git@68c51590da8a5097b7de6d75b4ccb6a175318b48",
     "us-equity-strategies @ git+https://github.com/QuantStrategyLab/UsEquityStrategies.git@b83ef4b3ae67c47d132ddd660ba3ccc60d474c85",
     "hk-equity-strategies @ git+https://github.com/QuantStrategyLab/HkEquityStrategies.git@709e5e1cde7841aed538d94eb26b552b46cb7806",
 ]
@@ -61,5 +61,5 @@ include = [
 
 [tool.uv]
 override-dependencies = [
-  "quant-platform-kit @ git+https://github.com/QuantStrategyLab/QuantPlatformKit.git@d51bb79dea8dc5773384d21b06e726620276558c",
+  "quant-platform-kit @ git+https://github.com/QuantStrategyLab/QuantPlatformKit.git@68c51590da8a5097b7de6d75b4ccb6a175318b48",
 ]

qsl.toml

@@ -5,7 +5,7 @@ upgrade_ring = "ring_d"
 allow_legacy = false
 
 [qsl.requires]
-quant_platform_kit = "d51bb79dea8dc5773384d21b06e726620276558c"
+quant_platform_kit = "68c51590da8a5097b7de6d75b4ccb6a175318b48"
 us_equity_strategies = "c31f28484b8489a1ecda8e595a6d2afd3bce8185"
 hk_equity_strategies = "709e5e1cde7841aed538d94eb26b552b46cb7806"