Merge pull request #279 from QuantStrategyLab/fix/retryable-funding-window

fix: safely retry pre-submission execution blocks

Author: Pigbibi

.github/workflows/sync-cloud-run-env.yml

@@ -754,6 +754,10 @@ jobs:
               --uri="${scheduler_uri}" \
               --schedule="${desired_schedule}" \
               --time-zone="${market_timezone}" \
+              --max-retry-attempts=3 \
+              --min-backoff=300s \
+              --max-backoff=1800s \
+              --max-doublings=2 \
               --quiet
           else
             echo "Creating Cloud Scheduler job ${job_name} schedule ${desired_schedule}, timezone ${market_timezone}, and URI ${scheduler_uri}."
@@ -767,6 +771,10 @@ jobs:
               --schedule="${desired_schedule}" \
               --time-zone="${market_timezone}" \
               --attempt-deadline=600s \
+              --max-retry-attempts=3 \
+              --min-backoff=300s \
+              --max-backoff=1800s \
+              --max-doublings=2 \
               --quiet
           fi
 

README.md

@@ -26,6 +26,15 @@ It is an execution layer, not a strategy research repository. Strategy logic com
 - Must keep credentials in GitHub Secrets, cloud secret stores, or the broker-specific secret system, never in Git.
 - Should start with dry-run or paper mode before any live order path is enabled.
 
+## Live retry boundary
+
+The runtime retries only a cycle that made **no** broker-order request: for
+example, a temporary quote failure or insufficient settled cash. It writes a
+durable create-only claim immediately before the first live broker request, so
+an accepted, rejected, pending, timed-out, or otherwise unknown broker request
+is never sent again automatically. Funding blocks notify once and can retry on
+the bounded scheduler backoff or the next run inside the strategy window.
+
 ## Direct vs snapshot-backed profiles
 
 Direct runtime profiles can usually run from market history or portfolio state. Snapshot-backed profiles need a current artifact bundle from the matching snapshot pipeline before this platform should execute them. The platform should not invent strategy eligibility; it should consume the status and artifacts published by the strategy and snapshot repositories.

README.zh-CN.md

@@ -26,6 +26,13 @@ FirstradePlatform 是 QuantStrategyLab 的实验性 Firstrade 执行平台。实
 - 凭据必须放在 GitHub Secrets、云密钥系统或券商专用密钥系统中，不能提交到 Git。
 - 任何 live 下单路径启用前，都应先从 dry-run 或 paper mode 开始。
 
+## 实盘重试边界
+
+运行时只会重试**从未向券商发出订单请求**的周期，例如暂时拿不到报价或可用现金不足。第一次
+真实券商请求前会写入持久化、仅创建一次的提交锁；因此已被券商受理、拒绝、待处理、超时或结果
+未知的请求都不会自动重发。资金不足只提醒一次，并会在受限的调度退避或策略窗口内的下一次运行时
+再次检查。
+
 ## 普通 profile 与 snapshot-backed profile
 
 普通 runtime profile 通常可以直接基于 market history 或 portfolio state 执行。Snapshot-backed profile 需要先从对应 snapshot pipeline 获取当前 artifact bundle，平台才应该执行。平台不应该自行判断策略资格，而应消费策略仓和 snapshot 仓发布的状态与产物。

application/execution_service.py

@@ -4,6 +4,7 @@
 
 from dataclasses import dataclass
 from math import floor
+from collections.abc import Callable
 from typing import Any
 
 from quant_platform_kit.common.order_status import compute_confirmed_sell_release_value
@@ -137,6 +138,7 @@ class ExecutionCycleResult:
     broker_submission_done: bool = False
     pending_reconciliation: bool = False
     execution_notes: tuple[dict[str, Any], ...] = ()
+    idempotency_blocked: bool = False
 
 
 DEFAULT_SAFE_HAVEN_CASH_SUBSTITUTE_THRESHOLD_USD = 1000.0
@@ -644,6 +646,7 @@ def execute_value_target_plan(
     cash_only_execution: bool = True,
     notional_buy_execution: bool = False,
     fetch_order_status=None,
+    before_live_submission: Callable[[], bool] | None = None,
 ) -> ExecutionCycleResult:
     del dry_run_only  # ExecutionPort owns preview vs live submission.
     plan = substitute_small_safe_haven_targets_with_cash(
@@ -702,6 +705,24 @@ def execute_value_target_plan(
     submitted_sell_orders: list[dict[str, Any]] = []
     sell_submitted = False
 
+    def _submission_claim_unavailable(symbol: str) -> ExecutionCycleResult:
+        skipped.append(
+            {
+                "symbol": symbol,
+                "reason": "duplicate_live_strategy_run",
+            }
+        )
+        return ExecutionCycleResult(
+            submitted_orders=tuple(submitted),
+            skipped_orders=tuple(skipped),
+            action_done=False,
+            execution_notes=tuple(execution_notes),
+            idempotency_blocked=True,
+        )
+
+    def _may_submit_live_order() -> bool:
+        return before_live_submission is None or bool(before_live_submission())
+
     tradable_deltas: list[tuple[str, float, float]] = []
     for symbol in sorted(set(targets) | set(market_values)):
         target_value = float(targets.get(symbol, 0.0))
@@ -749,6 +770,8 @@ def execute_value_target_plan(
                 )
                 continue
             sell_limit_price = price * float(limit_sell_discount)
+            if not _may_submit_live_order():
+                return _submission_claim_unavailable(symbol)
             order_result = _submit_order(
                 execution_port,
                 symbol=symbol,
@@ -855,6 +878,8 @@ def execute_value_target_plan(
                     }
                 )
                 continue
+            if not _may_submit_live_order():
+                return _submission_claim_unavailable(symbol)
             order_result = _submit_notional_buy_order(
                 execution_port,
                 symbol=symbol,
@@ -903,6 +928,8 @@ def execute_value_target_plan(
                     }
                 )
             continue
+        if not _may_submit_live_order():
+            return _submission_claim_unavailable(symbol)
         order_result = _submit_order(
             execution_port,
             symbol=symbol,

application/rebalance_service.py

@@ -31,8 +31,10 @@
 from application.strategy_run_persistence import (
     build_strategy_run_state,
     claim_live_strategy_run,
+    has_effective_live_submission_claim,
     is_duplicate_live_run,
     persist_strategy_run_state,
+    read_live_strategy_run_claim,
     read_latest_strategy_run_state,
     resolve_strategy_run_period,
 )
@@ -41,7 +43,7 @@
 from quant_platform_kit.common.execution_outcomes import (
     DEFAULT_EXECUTION_BLOCKING_SKIP_REASONS,
     filter_execution_blocking_skips,
-    is_terminal_funding_block,
+    is_funding_block,
     resolve_strategy_run_stage,
 )
 from quant_platform_kit.common.runtime_inputs import (
@@ -259,6 +261,8 @@ def send_and_capture(text: str) -> bool | None:
 
 
 def _should_publish_cycle_notification(result: Mapping[str, Any]) -> bool:
+    if result.get("notification_suppressed_by_policy"):
+        return False
     if result.get("submitted_orders"):
         return True
     if result.get("error") or result.get("ok") is False:
@@ -513,27 +517,22 @@ def log_message(message: str) -> None:
     masked_account = mask_account_id(account)
     existing_run = None
     if persist_strategy_runs and not settings.dry_run_only:
-        claim_acquired = claim_live_strategy_run(
+        existing_run = read_latest_strategy_run_state(
             store=store,
             account=masked_account,
             strategy_profile=strategy_runtime.profile,
             run_period=run_period,
-            now=now,
         )
-        existing_run = read_latest_strategy_run_state(
+        existing_submission_claim = read_live_strategy_run_claim(
             store=store,
             account=masked_account,
             strategy_profile=strategy_runtime.profile,
             run_period=run_period,
         )
-        if not claim_acquired and existing_run is None:
-            existing_run = {
-                "stage": "PENDING_SUBMISSION",
-                "as_of": now.isoformat(),
-                "claim_only": True,
-            }
-        if not claim_acquired or is_duplicate_live_run(existing_run):
-            duplicate_stage = str(existing_run.get("stage") or "NO_ACTION")
+        claim_blocks_repeat = has_effective_live_submission_claim(existing_submission_claim)
+        if claim_blocks_repeat or is_duplicate_live_run(existing_run):
+            existing_state = dict(existing_run or {})
+            duplicate_stage = str(existing_state.get("stage") or "PENDING_SUBMISSION")
             duplicate_skipped_orders = [
                 {
                     "reason": "duplicate_live_strategy_run",
@@ -559,17 +558,18 @@ def log_message(message: str) -> None:
                 now=now,
             )
             duplicate_state["idempotency_skipped"] = True
-            duplicate_state["existing_strategy_run_stage"] = existing_run.get("stage")
-            duplicate_state["existing_strategy_run_as_of"] = existing_run.get("as_of")
-            try:
-                strategy_run_persisted = persist_strategy_run_state(
-                    store=store,
-                    state=duplicate_state,
-                    now=now,
-                )
-            except Exception as exc:
-                strategy_run_persisted = False
-                strategy_run_persistence_error = f"{type(exc).__name__}: {exc}"
+            duplicate_state["existing_strategy_run_stage"] = existing_state.get("stage")
+            duplicate_state["existing_strategy_run_as_of"] = existing_state.get("as_of")
+            if not claim_blocks_repeat:
+                try:
+                    strategy_run_persisted = persist_strategy_run_state(
+                        store=store,
+                        state=duplicate_state,
+                        now=now,
+                    )
+                except Exception as exc:
+                    strategy_run_persisted = False
+                    strategy_run_persistence_error = f"{type(exc).__name__}: {exc}"
             result = {
                 "ok": True,
                 "api_kind": "unofficial-reverse-engineered",
@@ -583,8 +583,9 @@ def log_message(message: str) -> None:
                 "strategy_run_stage": duplicate_stage,
                 "strategy_run_persisted": strategy_run_persisted,
                 "idempotency_skipped": True,
-                "existing_strategy_run_stage": existing_run.get("stage"),
-                "existing_strategy_run_as_of": existing_run.get("as_of"),
+                "submission_claim_blocks_repeat": claim_blocks_repeat,
+                "existing_strategy_run_stage": existing_state.get("stage"),
+                "existing_strategy_run_as_of": existing_state.get("as_of"),
                 "submitted_orders": [],
                 "skipped_orders": duplicate_skipped_orders,
                 "action_done": False,
@@ -636,6 +637,22 @@ def log_message(message: str) -> None:
         except Exception as exc:
             strategy_run_persisted = False
             strategy_run_persistence_error = f"{type(exc).__name__}: {exc}"
+
+    submission_claim_acquired = False
+
+    def acquire_submission_claim() -> bool:
+        nonlocal submission_claim_acquired
+        if submission_claim_acquired:
+            return True
+        submission_claim_acquired = claim_live_strategy_run(
+            store=store,
+            account=masked_account,
+            strategy_profile=strategy_runtime.profile,
+            run_period=run_period,
+            now=now,
+        )
+        return submission_claim_acquired
+
     execution_result = execute_value_target_plan(
         plan=plan,
         market_data_port=market_data_port,
@@ -649,24 +666,80 @@ def log_message(message: str) -> None:
         cash_only_execution=settings.cash_only_execution,
         notional_buy_execution=notional_buy_execution_enabled(settings.strategy_profile),
         fetch_order_status=lambda broker_order_id: client.get_order_status(account, broker_order_id),
+        before_live_submission=(
+            acquire_submission_claim
+            if persist_strategy_runs and not settings.dry_run_only
+            else None
+        ),
     )
+    if execution_result.idempotency_blocked:
+        existing_run = read_latest_strategy_run_state(
+            store=store,
+            account=masked_account,
+            strategy_profile=strategy_runtime.profile,
+            run_period=run_period,
+        ) or {
+            "stage": "PENDING_SUBMISSION",
+            "as_of": now.isoformat(),
+            "claim_only": True,
+        }
+        result = {
+            "ok": True,
+            "api_kind": "unofficial-reverse-engineered",
+            "account": account,
+            "strategy_profile": strategy_runtime.profile,
+            "strategy_display_name": strategy_runtime.display_name,
+            "dry_run_only": settings.dry_run_only,
+            "live_trading_enabled": settings.live_trading_enabled,
+            "session_reused": bool(getattr(client, "session_reused", False)),
+            "strategy_run_period": run_period,
+            "strategy_run_stage": str(existing_run.get("stage") or "PENDING_SUBMISSION"),
+            "strategy_run_persisted": strategy_run_persisted,
+            "idempotency_skipped": True,
+            "existing_strategy_run_stage": existing_run.get("stage"),
+            "existing_strategy_run_as_of": existing_run.get("as_of"),
+            "submitted_orders": [],
+            "skipped_orders": list(execution_result.skipped_orders),
+            "action_done": False,
+            **empty_strategy_plugin_alert_report_fields(),
+        }
+        if strategy_run_persistence_error:
+            result["strategy_run_persistence_error"] = strategy_run_persistence_error
+        return attach_strategy_plugin_result(
+            result,
+            signals=strategy_plugin_signals,
+            error=strategy_plugin_error,
+            translator=translator,
+        )
     submitted_orders = list(execution_result.submitted_orders)
     skipped_orders = list(execution_result.skipped_orders)
     execution_notes = list(execution_result.execution_notes)
+    decision_diagnostics = dict(getattr(evaluation.decision, "diagnostics", {}) or {})
+    strategy_funding_shortfall = (
+        not submitted_orders
+        and str(
+            decision_diagnostics.get("dca_skip_reason")
+            or decision_diagnostics.get("skip_reason")
+            or ""
+        ).strip().lower()
+        == "insufficient_cash"
+    )
+    if strategy_funding_shortfall:
+        skipped_orders.append({"reason": "insufficient_cash"})
     blocking_skips = filter_execution_blocking_skips(
         skipped_orders,
         blocking_reasons=BROKER_EXECUTION_BLOCKING_SKIP_REASONS,
     )
     execution_blocked = bool(blocking_skips)
-    funding_blocked = is_terminal_funding_block(blocking_skips)
-    terminal_funding_block = funding_blocked and not execution_result.action_done
+    funding_blocked = is_funding_block(blocking_skips)
+    funding_block = funding_blocked and not execution_result.action_done
     strategy_run_stage = (
         "PENDING_RECONCILIATION"
         if execution_result.pending_reconciliation
         else resolve_strategy_run_stage(
             dry_run_only=settings.dry_run_only,
             execution_blocked=execution_blocked,
-            terminal_funding_block=terminal_funding_block,
+            terminal_funding_block=funding_block,
             action_done=execution_result.action_done,
         )
     )
@@ -712,11 +785,14 @@ def log_message(message: str) -> None:
     }
     if execution_blocked:
         result["execution_blocked"] = True
-        result["execution_block_retryable"] = not terminal_funding_block
+        result["execution_block_retryable"] = not submission_claim_acquired
         result["execution_blocking_skips"] = blocking_skips
         result["error"] = "Strategy execution blocked; see execution_blocking_skips."
     if funding_blocked:
         result["funding_blocked"] = True
+        if str((existing_run or {}).get("stage") or "").upper() == "FUNDING_BLOCKED":
+            result["notification_suppressed_by_policy"] = True
+            result["notification_suppressed_reason"] = "repeat_funding_blocked"
     if strategy_run_persistence_error:
         result["strategy_run_persistence_error"] = strategy_run_persistence_error
     if strategy_plugin_alert_result is not None:
@@ -744,9 +820,9 @@ def log_message(message: str) -> None:
             portfolio_snapshot=plan.get("portfolio", {}),
             evaluation_metadata=getattr(evaluation, "metadata", None),
             plan=plan,
-            submitted_orders=list(execution_result.submitted_orders),
-            skipped_orders=list(execution_result.skipped_orders),
-            execution_notes=list(execution_result.execution_notes),
+            submitted_orders=submitted_orders,
+            skipped_orders=skipped_orders,
+            execution_notes=execution_notes,
             action_done=execution_result.action_done,
             broker_submission_done=execution_result.broker_submission_done,
             execution_status=result["execution_status"],
@@ -777,7 +853,7 @@ def log_message(message: str) -> None:
     elif send_cycle_notification:
         result["notification_sent"] = False
         result["notification_suppressed"] = True
-        result["notification_suppressed_reason"] = "no_trade_or_error"
+        result.setdefault("notification_suppressed_reason", "no_trade_or_error")
     else:
         result["notification_sent"] = False
         result["notification_suppressed"] = True