From 771f18ff7accc55cefa2f1a7efb6fea7987585f7 Mon Sep 17 00:00:00 2001 From: Pigbibi <20649888+Pigbibi@users.noreply.github.com> Date: Sun, 30 Aug 2026 18:31:18 +0800 Subject: [PATCH] fix: safely retry pre-submission execution blocks Co-Authored-By: Codex --- .github/workflows/sync-cloud-run-env.yml | 8 ++ README.md | 9 ++ README.zh-CN.md | 7 ++ application/execution_service.py | 27 +++++ application/rebalance_service.py | 142 +++++++++++++++++------ application/strategy_run_persistence.py | 32 +++++ main.py | 2 +- notifications/telegram.py | 4 +- pyproject.toml | 4 +- tests/test_execution_service.py | 31 +++++ tests/test_rebalance_service.py | 83 ++++++++++--- tests/test_request_handling.py | 8 +- tests/test_strategy_run_claim.py | 11 ++ uv.lock | 6 +- 14 files changed, 313 insertions(+), 61 deletions(-) diff --git a/.github/workflows/sync-cloud-run-env.yml b/.github/workflows/sync-cloud-run-env.yml index 0c60ec3..27867ab 100644 --- a/.github/workflows/sync-cloud-run-env.yml +++ b/.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 diff --git a/README.md b/README.md index 52c1b0b..c93211b 100644 --- a/README.md +++ b/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. diff --git a/README.zh-CN.md b/README.zh-CN.md index 3052119..16addb9 100644 --- a/README.zh-CN.md +++ b/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 仓发布的状态与产物。 diff --git a/application/execution_service.py b/application/execution_service.py index e691c7e..481ea80 100644 --- a/application/execution_service.py +++ b/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, diff --git a/application/rebalance_service.py b/application/rebalance_service.py index dc2f9e2..46dc609 100644 --- a/application/rebalance_service.py +++ b/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 diff --git a/application/strategy_run_persistence.py b/application/strategy_run_persistence.py index 5bbbf91..ed9dad0 100644 --- a/application/strategy_run_persistence.py +++ b/application/strategy_run_persistence.py @@ -21,6 +21,7 @@ STAGE_NO_ACTION, } ) +LIVE_SUBMISSION_CLAIM_BOUNDARY = "pre_broker_request_v2" def utcnow() -> datetime: @@ -133,6 +134,7 @@ def claim_live_strategy_run( """Acquire the durable pre-order claim using object-store create-if-absent.""" payload = { "stage": "PENDING_SUBMISSION", + "submission_claim_boundary": LIVE_SUBMISSION_CLAIM_BOUNDARY, "account": account, "strategy_profile": strategy_profile, "run_period": run_period, @@ -147,6 +149,36 @@ def claim_live_strategy_run( ) +def read_live_strategy_run_claim( + *, + store: GcsStateStore, + account: str, + strategy_profile: str, + run_period: str, +) -> dict[str, Any] | None: + return store.read_json( + strategy_run_claim_key( + account=account, + strategy_profile=strategy_profile, + run_period=run_period, + ) + ) + + +def has_effective_live_submission_claim(claim: Mapping[str, Any] | None) -> bool: + """Return whether a claim was created at the current pre-order boundary. + + Claims written before this boundary existed were acquired too early and do + not prove a broker request. Keeping them non-blocking avoids carrying a + historical no-order lock into the retryable lifecycle. + """ + + return ( + str((claim or {}).get("submission_claim_boundary") or "").strip() + == LIVE_SUBMISSION_CLAIM_BOUNDARY + ) + + def read_latest_strategy_run_state( *, store: GcsStateStore, diff --git a/main.py b/main.py index b8342d6..62e86f3 100644 --- a/main.py +++ b/main.py @@ -295,7 +295,7 @@ def _strategy_result_diagnostics(result: dict[str, Any]) -> dict[str, Any]: def _strategy_result_http_status(result: dict[str, Any]) -> int: - if result.get("execution_blocked") and result.get("execution_block_retryable") and not result.get("funding_blocked"): + if result.get("execution_blocked") and result.get("execution_block_retryable"): return 500 return 200 diff --git a/notifications/telegram.py b/notifications/telegram.py index 97c66f9..77f1c55 100644 --- a/notifications/telegram.py +++ b/notifications/telegram.py @@ -245,7 +245,7 @@ def format_small_account_whole_share_bootstrap_notes( "no_order_submitted": "未下单: 原因={reason}", "execution_blocked_banner": "⚠️ 执行阻塞: {reason}", "execution_blocked_retryable_banner": "⚠️ 执行阻塞，可在窗口内自动重试: {reason}", - "funding_blocked_banner": "⚠️ 资金不足，本周期不再自动重试: {reason}", + "funding_blocked_banner": "⚠️ 资金不足，已提醒；资金到账后将在窗口内自动重试: {reason}", "no_rebalance_needed": "✅ 无需调仓", "no_trades": "✅ 无需调仓", "no_executable_orders": "无可执行订单", @@ -418,7 +418,7 @@ def format_small_account_whole_share_bootstrap_notes( "no_order_submitted": "No order submitted: reason={reason}", "execution_blocked_banner": "⚠️ Execution blocked: {reason}", "execution_blocked_retryable_banner": "⚠️ Execution blocked; retryable within window: {reason}", - "funding_blocked_banner": "⚠️ Funding blocked; no more automatic retries for this period: {reason}", + "funding_blocked_banner": "⚠️ Funding blocked; alerted once and will retry within the window after funds arrive: {reason}", "no_rebalance_needed": "✅ No rebalance needed", "no_trades": "✅ No rebalance needed", "no_executable_orders": "no executable orders", diff --git a/pyproject.toml b/pyproject.toml index d1c4442..fd52ba0 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -18,7 +18,7 @@ dependencies = [ "pytest", "pytz", "requests", - "quant-platform-kit @ git+https://github.com/QuantStrategyLab/QuantPlatformKit.git@cc8891a66cf58d9d339f5d257b412f3f7608a149", + "quant-platform-kit @ git+https://github.com/QuantStrategyLab/QuantPlatformKit.git@2d572658df6e1e82ce822e782e4a717beda4cf83", "us-equity-strategies @ git+https://github.com/QuantStrategyLab/UsEquityStrategies.git@26c9b908c486ca60123414eda42cd509c0f17c62", ] license = "MIT" @@ -82,5 +82,5 @@ show_missing = true [tool.uv] override-dependencies = [ - "quant-platform-kit @ git+https://github.com/QuantStrategyLab/QuantPlatformKit.git@cc8891a66cf58d9d339f5d257b412f3f7608a149", + "quant-platform-kit @ git+https://github.com/QuantStrategyLab/QuantPlatformKit.git@2d572658df6e1e82ce822e782e4a717beda4cf83", ] diff --git a/tests/test_execution_service.py b/tests/test_execution_service.py index 51869fd..506a3b9 100644 --- a/tests/test_execution_service.py +++ b/tests/test_execution_service.py @@ -107,6 +107,37 @@ def test_execute_value_target_plan_marks_live_submissions_pending_reconciliation assert len(result.submitted_orders) == 1 +def test_execute_value_target_plan_stops_before_broker_call_when_submission_claim_is_taken(): + execution_port = FakeExecutionPort() + claim_attempts = [] + + def reject_submission_claim() -> bool: + claim_attempts.append(True) + return False + + result = execute_value_target_plan( + plan={ + "allocation": {"targets": {"AAA": 20.0}}, + "portfolio": { + "market_values": {"AAA": 0.0}, + "sellable_quantities": {"AAA": 0.0}, + "liquid_cash": 100.0, + }, + "execution": {"current_min_trade": 5.0, "investable_cash": 100.0}, + }, + market_data_port=FakeMarketDataPort({"AAA": 10.0}), + execution_port=execution_port, + dry_run_only=False, + before_live_submission=reject_submission_claim, + ) + + assert claim_attempts == [True] + assert execution_port.orders == [] + assert result.idempotency_blocked is True + assert result.submitted_orders == () + assert result.skipped_orders == ({"symbol": "AAA", "reason": "duplicate_live_strategy_run"},) + + def test_execute_value_target_plan_uses_sellable_quantity_when_market_value_is_stale_below_quote(): execution_port = FakeExecutionPort() result = execute_value_target_plan( diff --git a/tests/test_rebalance_service.py b/tests/test_rebalance_service.py index 550a2e9..62dafe4 100644 --- a/tests/test_rebalance_service.py +++ b/tests/test_rebalance_service.py @@ -780,7 +780,7 @@ def test_run_strategy_cycle_persists_live_execution_blocked_without_terminal_sta assert latest_payload["stage"] == "EXECUTION_BLOCKED" -def test_run_strategy_cycle_persists_live_funding_block_as_terminal(monkeypatch): +def test_run_strategy_cycle_retries_live_funding_block_after_cash_arrives(monkeypatch): store = FakeStateStore() settings = _runtime_settings_with_persistence( dry_run_only=False, @@ -790,9 +790,17 @@ def test_run_strategy_cycle_persists_live_funding_block_as_terminal(monkeypatch) max_order_notional_usd=None, ) + available_cash = {"value": "50.00"} + observed_clients = [] + messages = [] + class FundingBlockedClient(FakeFirstradeClient): def get_balances(self, _account): - return {"total_value": "150.00", "cash": "50.00", "buying_power": "50.00"} + return { + "total_value": "150.00", + "cash": available_cash["value"], + "buying_power": available_cash["value"], + } def get_quote(self, _account, symbol): return {"symbol": symbol, "last": "100.00", "bid": "99.90", "ask": "100.10"} @@ -815,11 +823,17 @@ def evaluate(self, **inputs): lambda *_args, **_kwargs: FundingBlockedRuntime(), ) + def funding_blocked_client_factory(*args, **kwargs): + client = FundingBlockedClient(*args, **kwargs) + observed_clients.append(client) + return client + result = run_strategy_cycle( runtime_settings=settings, credentials=FirstradeCredentials(username="user", password="pass"), - client_factory=FundingBlockedClient, + client_factory=funding_blocked_client_factory, state_store=store, + notification_sender=messages.append, env_reader=lambda _name, default=None: default, ) @@ -827,32 +841,69 @@ def evaluate(self, **inputs): assert result["action_done"] is False assert result["ok"] is False assert result["execution_blocked"] is True - assert result["execution_block_retryable"] is False + assert result["execution_block_retryable"] is True assert result["funding_blocked"] is True assert result["strategy_run_stage"] == "FUNDING_BLOCKED" assert result["skipped_orders"][0]["reason"] == "insufficient_cash_for_whole_share" assert latest_payload["stage"] == "FUNDING_BLOCKED" + assert not [key for key in store.payloads if key.startswith("strategy-runs/claims/")] + assert observed_clients[0].orders == [] + assert result["notification_sent"] is True + assert len(messages) == 1 write_count = len(store.writes) + repeated_funding_result = run_strategy_cycle( + runtime_settings=settings, + credentials=FirstradeCredentials(username="user", password="pass"), + client_factory=funding_blocked_client_factory, + state_store=store, + notification_sender=messages.append, + env_reader=lambda _name, default=None: default, + ) + + assert repeated_funding_result["funding_blocked"] is True + assert repeated_funding_result["notification_suppressed_reason"] == "repeat_funding_blocked" + assert repeated_funding_result["notification_sent"] is False + assert len(messages) == 1 + + available_cash["value"] = "150.00" second_result = run_strategy_cycle( runtime_settings=settings, credentials=FirstradeCredentials(username="user", password="pass"), - client_factory=FundingBlockedClient, + client_factory=funding_blocked_client_factory, state_store=store, + notification_sender=messages.append, env_reader=lambda _name, default=None: default, ) - assert second_result["idempotency_skipped"] is True - assert second_result["existing_strategy_run_stage"] == "FUNDING_BLOCKED" - assert second_result["strategy_run_stage"] == "FUNDING_BLOCKED" + assert second_result.get("idempotency_skipped") is not True + assert second_result["broker_submission_done"] is True + assert second_result["strategy_run_stage"] == "PENDING_RECONCILIATION" assert second_result["strategy_run_persisted"] is True - assert len(store.writes) == write_count + 2 + assert len(store.writes) == write_count + 8 + assert second_result["submitted_orders"][0]["symbol"] == "AAA" + assert observed_clients[2].orders[0][1] is False + claim_keys = [key for key in store.payloads if key.startswith("strategy-runs/claims/")] + assert len(claim_keys) == 1 latest_payloads = _latest_strategy_run_payloads(store) - duplicate_payload = latest_payloads[-1] - assert duplicate_payload["stage"] == "FUNDING_BLOCKED" - assert duplicate_payload["idempotency_skipped"] is True - assert duplicate_payload["existing_strategy_run_stage"] == "FUNDING_BLOCKED" - assert duplicate_payload["skipped_orders"][0]["reason"] == "duplicate_live_strategy_run" + assert latest_payloads[-1]["stage"] == "PENDING_RECONCILIATION" + assert len(messages) == 2 + + write_count_after_submission = len(store.writes) + after_submission_result = run_strategy_cycle( + runtime_settings=settings, + credentials=FirstradeCredentials(username="user", password="pass"), + client_factory=funding_blocked_client_factory, + state_store=store, + notification_sender=messages.append, + env_reader=lambda _name, default=None: default, + ) + + assert after_submission_result["idempotency_skipped"] is True + assert after_submission_result["submission_claim_blocks_repeat"] is True + assert after_submission_result["strategy_run_stage"] == "PENDING_RECONCILIATION" + assert observed_clients[3].orders == [] + assert len(store.writes) == write_count_after_submission def test_run_strategy_cycle_persists_live_partial_submission_as_non_terminal(monkeypatch): @@ -1146,7 +1197,7 @@ def test_render_cycle_summary_shows_funding_blocked_banner(): "strategy_display_name": "Russell Top50 Leader Rotation", "dry_run_only": False, "execution_blocked": True, - "execution_block_retryable": False, + "execution_block_retryable": True, "funding_blocked": True, "execution_blocking_skips": [ {"symbol": "NVDA", "reason": "insufficient_cash_for_whole_share"} @@ -1168,7 +1219,7 @@ def test_render_cycle_summary_shows_funding_blocked_banner(): lang="zh", ) - assert "⚠️ 资金不足，本周期不再自动重试: NVDA（现金不足以买入一整股）" in message + assert "⚠️ 资金不足，已提醒；资金到账后将在窗口内自动重试: NVDA（现金不足以买入一整股）" in message def test_render_cycle_summary_shows_retryable_execution_blocked_banner(): diff --git a/tests/test_request_handling.py b/tests/test_request_handling.py index d5412e3..ebcd3e0 100644 --- a/tests/test_request_handling.py +++ b/tests/test_request_handling.py @@ -138,7 +138,7 @@ def test_run_endpoint_skips_when_market_closed(monkeypatch): assert payload["submitted_orders"] == [] -def test_run_endpoint_returns_200_for_terminal_funding_block(monkeypatch): +def test_run_endpoint_returns_500_for_retryable_funding_block(monkeypatch): monkeypatch.setenv("FIRSTRADE_RUN_STRATEGY_ON_HTTP", "true") monkeypatch.setattr(main, "_should_skip_for_market_hours", lambda: (False, None)) monkeypatch.setattr( @@ -147,7 +147,7 @@ def test_run_endpoint_returns_200_for_terminal_funding_block(monkeypatch): lambda **_kwargs: { "ok": False, "execution_blocked": True, - "execution_block_retryable": False, + "execution_block_retryable": True, "funding_blocked": True, "error": "Strategy execution blocked; see execution_blocking_skips.", }, @@ -156,10 +156,10 @@ def test_run_endpoint_returns_200_for_terminal_funding_block(monkeypatch): response = client.post("/run") - assert response.status_code == 200 + assert response.status_code == 500 payload = response.get_json() assert payload["funding_blocked"] is True - assert payload["execution_block_retryable"] is False + assert payload["execution_block_retryable"] is True def test_run_endpoint_returns_500_for_retryable_execution_block(monkeypatch): diff --git a/tests/test_strategy_run_claim.py b/tests/test_strategy_run_claim.py index 40d3f67..9e2353b 100644 --- a/tests/test_strategy_run_claim.py +++ b/tests/test_strategy_run_claim.py @@ -1,5 +1,7 @@ from application.strategy_run_persistence import ( claim_live_strategy_run, + has_effective_live_submission_claim, + read_live_strategy_run_claim, strategy_run_claim_key, ) from application.state_persistence import GcsStateStore @@ -15,6 +17,9 @@ def create_json(self, key, payload): self.payloads[key] = dict(payload) return True + def read_json(self, key): + return self.payloads.get(key) + def test_live_claim_is_create_only_and_permanent(): store = AtomicFakeStore() @@ -32,6 +37,12 @@ def test_live_claim_is_create_only_and_permanent(): ) assert store.payloads[key]["stage"] == "PENDING_SUBMISSION" assert store.payloads[key]["no_order_submitted"] is True + assert has_effective_live_submission_claim(store.payloads[key]) is True + assert read_live_strategy_run_claim(**kwargs) == store.payloads[key] + + +def test_legacy_early_claim_is_not_an_effective_submission_boundary(): + assert has_effective_live_submission_claim({"stage": "PENDING_SUBMISSION"}) is False def test_gcs_state_store_create_json_uses_generation_zero_precondition(): diff --git a/uv.lock b/uv.lock index 6b5572a..2303203 100644 --- a/uv.lock +++ b/uv.lock @@ -17,7 +17,7 @@ resolution-markers = [ ] [manifest] -overrides = [{ name = "quant-platform-kit", git = "https://github.com/QuantStrategyLab/QuantPlatformKit.git?rev=cc8891a66cf58d9d339f5d257b412f3f7608a149" }] +overrides = [{ name = "quant-platform-kit", git = "https://github.com/QuantStrategyLab/QuantPlatformKit.git?rev=2d572658df6e1e82ce822e782e4a717beda4cf83" }] [[package]] name = "blinker" @@ -446,7 +446,7 @@ requires-dist = [ { name = "pytest", marker = "extra == 'test'" }, { name = "pytest-cov", marker = "extra == 'test'" }, { name = "pytz" }, - { name = "quant-platform-kit", git = "https://github.com/QuantStrategyLab/QuantPlatformKit.git?rev=cc8891a66cf58d9d339f5d257b412f3f7608a149" }, + { name = "quant-platform-kit", git = "https://github.com/QuantStrategyLab/QuantPlatformKit.git?rev=2d572658df6e1e82ce822e782e4a717beda4cf83" }, { name = "requests" }, { name = "ruff", marker = "extra == 'test'" }, { name = "us-equity-strategies", git = "https://github.com/QuantStrategyLab/UsEquityStrategies.git?rev=26c9b908c486ca60123414eda42cd509c0f17c62" }, @@ -1102,7 +1102,7 @@ wheels = [ [[package]] name = "quant-platform-kit" version = "0.10.0" -source = { git = "https://github.com/QuantStrategyLab/QuantPlatformKit.git?rev=cc8891a66cf58d9d339f5d257b412f3f7608a149#cc8891a66cf58d9d339f5d257b412f3f7608a149" } +source = { git = "https://github.com/QuantStrategyLab/QuantPlatformKit.git?rev=2d572658df6e1e82ce822e782e4a717beda4cf83#2d572658df6e1e82ce822e782e4a717beda4cf83" } [[package]] name = "requests"