diff --git a/.github/workflows/sync-cloud-run-env.yml b/.github/workflows/sync-cloud-run-env.yml index 9c4ba9f..76876d4 100644 --- a/.github/workflows/sync-cloud-run-env.yml +++ b/.github/workflows/sync-cloud-run-env.yml @@ -15,6 +15,7 @@ on: options: - configured - hk-verify + - paper-command-verify cloud_run_region: description: "Cloud Run region for hk-verify. Leave blank to use the longbridge-hk Environment value." required: false @@ -49,6 +50,39 @@ on: required: true type: boolean default: true + paper_command_verify_service: + description: "Dedicated Cloud Run service for the paper command consumer." + required: false + type: string + default: longbridge-quant-paper-command-verify-service + paper_command_verify_strategy_profile: + description: "Strategy profile to verify with immutable paper commands." + required: false + type: string + default: russell_top50_leader_rotation + paper_command_verify_execution_command_cloud_uri: + description: "Dedicated GCS prefix for immutable paper commands; never an execution-report URI." + required: false + type: string + paper_command_verify_strategy_release_json: + description: "Complete strategy_release identity JSON required for paper command verification." + required: false + type: string + paper_command_verify_longport_secret_name: + description: "LongPort token secret name for the isolated paper command verifier." + required: false + type: string + default: longport_token_paper + paper_command_verify_longport_app_key_secret_name: + description: "LongPort app-key secret name for the isolated paper command verifier." + required: false + type: string + default: longport-app-key-paper + paper_command_verify_longport_app_secret_secret_name: + description: "LongPort app-secret secret name for the isolated paper command verifier." + required: false + type: string + default: longport-app-secret-paper env: GCP_PROJECT_ID: longbridgequant @@ -109,6 +143,13 @@ jobs: INPUT_LONGPORT_APP_SECRET_SECRET_NAME: ${{ inputs.longport_app_secret_secret_name }} INPUT_DEPLOY_IMAGE: ${{ inputs.deploy_image }} INPUT_SYNC_ENV: ${{ inputs.sync_env }} + INPUT_PAPER_COMMAND_VERIFY_SERVICE: ${{ inputs.paper_command_verify_service }} + INPUT_PAPER_COMMAND_VERIFY_STRATEGY_PROFILE: ${{ inputs.paper_command_verify_strategy_profile }} + INPUT_PAPER_COMMAND_VERIFY_EXECUTION_COMMAND_CLOUD_URI: ${{ inputs.paper_command_verify_execution_command_cloud_uri }} + INPUT_PAPER_COMMAND_VERIFY_STRATEGY_RELEASE_JSON: ${{ inputs.paper_command_verify_strategy_release_json }} + INPUT_PAPER_COMMAND_VERIFY_LONGPORT_SECRET_NAME: ${{ inputs.paper_command_verify_longport_secret_name }} + INPUT_PAPER_COMMAND_VERIFY_LONGPORT_APP_KEY_SECRET_NAME: ${{ inputs.paper_command_verify_longport_app_key_secret_name }} + INPUT_PAPER_COMMAND_VERIFY_LONGPORT_APP_SECRET_SECRET_NAME: ${{ inputs.paper_command_verify_longport_app_secret_secret_name }} GCP_ARTIFACT_REGISTRY_HOSTNAME: ${{ vars.GCP_ARTIFACT_REGISTRY_HOSTNAME }} # Set CLOUD_RUN_REGION per Environment so paper/HK/SG can target different regions. CLOUD_RUN_REGION: ${{ vars.CLOUD_RUN_REGION }} @@ -201,6 +242,7 @@ jobs: # Paper-only durable-command staging. These are intentionally opt-in; # a non-paper runtime rejects the producer even if configured. LONGBRIDGE_DURABLE_EXECUTION_COMMAND_PAPER_ENABLED: ${{ vars.LONGBRIDGE_DURABLE_EXECUTION_COMMAND_PAPER_ENABLED }} + LONGBRIDGE_DURABLE_EXECUTION_COMMAND_PAPER_CONSUMER_ENABLED: ${{ vars.LONGBRIDGE_DURABLE_EXECUTION_COMMAND_PAPER_CONSUMER_ENABLED }} LONGBRIDGE_EXECUTION_COMMAND_CLOUD_URI: ${{ vars.LONGBRIDGE_EXECUTION_COMMAND_CLOUD_URI }} GLOBAL_TELEGRAM_CHAT_ID: ${{ vars.GLOBAL_TELEGRAM_CHAT_ID }} TELEGRAM_TOKEN: ${{ secrets.TELEGRAM_TOKEN }} @@ -221,6 +263,7 @@ jobs: deploy_enabled=false env_sync_enabled=false + scheduler_sync_enabled=false # QSL_ENABLE_CLOUD_RUN_AUTOMATION overrides ENABLE_MAIN_PUSH_CLOUD_RUN_AUTOMATION ENABLE_MAIN_PUSH_CLOUD_RUN_AUTOMATION="${QSL_ENABLE_CLOUD_RUN_AUTOMATION:-$ENABLE_MAIN_PUSH_CLOUD_RUN_AUTOMATION}" @@ -244,7 +287,20 @@ jobs: exit 0 fi - if [ "${GITHUB_EVENT_NAME:-}" = "workflow_dispatch" ] && [ "${WORKFLOW_TARGET:-configured}" = "hk-verify" ]; then + if [ "${GITHUB_EVENT_NAME:-}" = "workflow_dispatch" ] \ + && [ "${WORKFLOW_TARGET:-configured}" = "paper-command-verify" ] \ + && [ "${DEPLOYMENT_LABEL:-}" != "PAPER" ]; then + write_github_output \ + "deploy_enabled=false" \ + "env_sync_enabled=false" \ + "scheduler_sync_enabled=false" \ + "enabled=false" + echo "Skipping ${DEPLOYMENT_LABEL} Cloud Run automation because paper-command-verify targets only the PAPER deployment." >&2 + exit 0 + fi + + if [ "${GITHUB_EVENT_NAME:-}" = "workflow_dispatch" ] \ + && { [ "${WORKFLOW_TARGET:-configured}" = "hk-verify" ] || [ "${WORKFLOW_TARGET:-configured}" = "paper-command-verify" ]; }; then if [ "${INPUT_DEPLOY_IMAGE:-true}" = "true" ]; then deploy_enabled=true fi @@ -261,9 +317,15 @@ jobs: fi fi + if [ "${env_sync_enabled}" = "true" ] \ + && [ "${WORKFLOW_TARGET:-configured}" != "paper-command-verify" ]; then + scheduler_sync_enabled=true + fi + write_github_output \ "deploy_enabled=${deploy_enabled}" \ - "env_sync_enabled=${env_sync_enabled}" + "env_sync_enabled=${env_sync_enabled}" \ + "scheduler_sync_enabled=${scheduler_sync_enabled}" if [ "${deploy_enabled}" != "true" ] && [ "${env_sync_enabled}" != "true" ]; then write_github_output "enabled=false" @@ -336,6 +398,94 @@ jobs: echo "CLOUD_RUN_ENV_SYNC_WAIT_FOR_COMMIT=false" >> "$GITHUB_ENV" fi + - name: Apply isolated paper-command verification defaults + if: steps.config.outputs.enabled == 'true' && github.event_name == 'workflow_dispatch' && inputs.target == 'paper-command-verify' + run: | + set -euo pipefail + + service="${INPUT_PAPER_COMMAND_VERIFY_SERVICE:-longbridge-quant-paper-command-verify-service}" + profile="${INPUT_PAPER_COMMAND_VERIFY_STRATEGY_PROFILE:-}" + command_uri="${INPUT_PAPER_COMMAND_VERIFY_EXECUTION_COMMAND_CLOUD_URI:-}" + release_json="${INPUT_PAPER_COMMAND_VERIFY_STRATEGY_RELEASE_JSON:-}" + if [ -z "${profile}" ] || [ -z "${command_uri}" ] || [ -z "${release_json}" ]; then + echo "paper-command-verify requires a strategy profile, a dedicated command GCS URI, and a complete strategy release identity." >&2 + exit 1 + fi + case "${command_uri}" in + gs://*) ;; + *) + echo "paper-command-verify command storage must be a gs:// URI." >&2 + exit 1 + ;; + esac + report_uri="${EXECUTION_REPORT_GCS_URI%/}" + command_uri_trimmed="${command_uri%/}" + if [ -n "${report_uri}" ] \ + && { [ "${command_uri_trimmed}" = "${report_uri}" ] || [[ "${command_uri_trimmed}" == "${report_uri}/"* ]]; }; then + echo "paper-command-verify command storage must not reuse the execution report URI or its prefix." >&2 + exit 1 + fi + + runtime_target="$(SERVICE="${service}" PROFILE="${profile}" RELEASE_JSON="${release_json}" python3 - <<'PY' + import json + import os + + release = json.loads(os.environ["RELEASE_JSON"]) + required = { + "release_id", + "manifest_sha256", + "strategy_revision", + "config_sha256", + "risk_policy_sha256", + "evidence_sha256", + "plugin_bundle_sha256", + "effective_session", + } + if not isinstance(release, dict) or required.difference(release): + raise SystemExit("paper-command-verify strategy release identity is incomplete") + print( + json.dumps( + { + "platform_id": "longbridge", + "strategy_profile": os.environ["PROFILE"], + "deployment_selector": "paper-command-verify", + "account_scope": "paper-command-verify", + "execution_mode": "paper", + "dry_run_only": True, + "service_name": os.environ["SERVICE"], + "strategy_release": release, + }, + separators=(",", ":"), + ) + ) + PY + )" + + { + echo "CLOUD_RUN_SERVICE=${service}" + echo "ACCOUNT_REGION=PAPER" + echo "ACCOUNT_PREFIX=PAPER_VERIFY" + echo "STRATEGY_PROFILE=${profile}" + echo "RUNTIME_TARGET_JSON=${runtime_target}" + echo "RUNTIME_TARGET_ENABLED=false" + echo "LONGBRIDGE_DRY_RUN_ONLY=true" + echo "LONGBRIDGE_MARKET=US" + echo "LONGBRIDGE_MARKET_CALENDAR=NYSE" + echo "LONGBRIDGE_MARKET_TIMEZONE=America/New_York" + echo "LONGBRIDGE_SYMBOL_SUFFIX=.US" + echo "LONGBRIDGE_TRADING_CURRENCY=USD" + echo "LONGBRIDGE_DURABLE_EXECUTION_COMMAND_PAPER_ENABLED=true" + echo "LONGBRIDGE_DURABLE_EXECUTION_COMMAND_PAPER_CONSUMER_ENABLED=true" + echo "LONGBRIDGE_EXECUTION_COMMAND_CLOUD_URI=${command_uri_trimmed}" + echo "LONGPORT_SECRET_NAME=${INPUT_PAPER_COMMAND_VERIFY_LONGPORT_SECRET_NAME}" + echo "LONGPORT_APP_KEY_SECRET_NAME=${INPUT_PAPER_COMMAND_VERIFY_LONGPORT_APP_KEY_SECRET_NAME}" + echo "LONGPORT_APP_SECRET_SECRET_NAME=${INPUT_PAPER_COMMAND_VERIFY_LONGPORT_APP_SECRET_SECRET_NAME}" + } >> "$GITHUB_ENV" + + if [ "${INPUT_DEPLOY_IMAGE:-true}" != "true" ]; then + echo "CLOUD_RUN_ENV_SYNC_WAIT_FOR_COMMIT=false" >> "$GITHUB_ENV" + fi + - name: Set up Python for strategy requirement resolution if: steps.config.outputs.env_sync_enabled == 'true' uses: actions/setup-python@v6 @@ -902,7 +1052,7 @@ jobs: python3 scripts/reconcile_cloud_runtime.py --platform longbridge --ensure-latest-traffic --service "${CLOUD_RUN_SERVICE}" - name: Sync Cloud Scheduler schedule - if: steps.config.outputs.env_sync_enabled == 'true' + if: steps.config.outputs.scheduler_sync_enabled == 'true' env: SYNC_PLAN_JSON: ${{ steps.strategy_requirements.outputs.sync_plan_json }} run: | @@ -1186,7 +1336,7 @@ jobs: done - name: Reconcile legacy Cloud Scheduler jobs - if: steps.config.outputs.env_sync_enabled == 'true' + if: steps.config.outputs.scheduler_sync_enabled == 'true' env: SYNC_PLAN_JSON: ${{ steps.strategy_requirements.outputs.sync_plan_json }} run: | diff --git a/application/durable_execution_commands.py b/application/durable_execution_commands.py index b525c11..13202bf 100644 --- a/application/durable_execution_commands.py +++ b/application/durable_execution_commands.py @@ -18,6 +18,7 @@ RuntimeCommandGatePolicy, evaluate_runtime_command_gate, ) +from quant_platform_kit.common.strategy_release import build_strategy_release_identity PAPER_EXECUTION_INTENT_SCHEMA_VERSION = "longbridge.paper-execution-intent.v1" @@ -55,6 +56,7 @@ def build_paper_execution_command( strategy_profile: str, execution: Mapping[str, Any], allocation: Mapping[str, Any], + strategy_release: Any = None, ) -> ExecutionCommand: """Bind one paper-only command to immutable timing and target intent.""" execution = dict(execution or {}) @@ -67,6 +69,12 @@ def build_paper_execution_command( "risk_symbols": _normalized_symbols(allocation.get("risk_symbols")), "safe_haven_symbols": _normalized_symbols(allocation.get("safe_haven_symbols")), } + if strategy_release is not None: + # The command is content-addressed, so including the release identity + # binds a delayed paper command to the exact decision release. The + # future consumer compares it with its self-attested runtime release + # before it simulates even a single order. + intent["strategy_release"] = build_strategy_release_identity(strategy_release).to_dict() intent_json = _canonical_json(intent) return ExecutionCommand.from_decision( platform=platform, @@ -107,6 +115,7 @@ def enqueue_paper_execution_command( strategy_profile=strategy_profile, execution=execution, allocation=allocation, + strategy_release=expected_strategy_release, ) created = store.enqueue(command) gate_decision = evaluate_runtime_command_gate( @@ -153,3 +162,14 @@ def resolve_paper_execution_command_producer_enabled(*, env_reader, dry_run_only if enabled and not dry_run_only: raise RuntimeError("durable execution command producer is paper-only and cannot be enabled live") return enabled + + +def resolve_paper_execution_command_consumer_enabled(*, env_reader, dry_run_only: bool) -> bool: + """Resolve the opt-in paper consumer flag and reject any live runtime.""" + raw_value = str( + env_reader("LONGBRIDGE_DURABLE_EXECUTION_COMMAND_PAPER_CONSUMER_ENABLED", "") or "" + ).strip().lower() + enabled = raw_value in {"1", "true", "t", "yes", "y", "on"} + if enabled and not dry_run_only: + raise RuntimeError("durable execution command consumer is paper-only and cannot be enabled live") + return enabled diff --git a/application/paper_execution_command_consumer.py b/application/paper_execution_command_consumer.py new file mode 100644 index 0000000..8f7ae7a --- /dev/null +++ b/application/paper_execution_command_consumer.py @@ -0,0 +1,437 @@ +"""Paper-only durable-command consumer with reconciled per-symbol evidence. + +The module intentionally never imports an execution adapter. It reads the +current account snapshot and quotes, records what *would* be submitted, and +uses the shared runtime command gate in observation mode. A command is only +marked as paper-filled when every reconciled proposal passes the policy; no +broker order is ever created here. +""" + +from __future__ import annotations + +import math +from collections.abc import Mapping +from datetime import date +from typing import Any + +from quant_platform_kit.common.execution_commands import ( + ExecutionCommand, + ExecutionCommandState, + ExecutionCommandStore, +) +from quant_platform_kit.common.runtime_command_gate import ( + RuntimeCommandAction, + RuntimeCommandExposureEffect, + RuntimeCommandGateEnforcement, + RuntimeCommandGatePolicy, + evaluate_runtime_command_gate, +) +from quant_platform_kit.common.strategy_release import ( + StrategyReleaseIdentity, + build_strategy_release_identity, +) + + +PAPER_COMMAND_CONSUMER_SCHEMA_VERSION = "longbridge.paper-execution-command-consumer.v1" +PAPER_EXECUTION_INTENT_SCHEMA_VERSION = "longbridge.paper-execution-intent.v1" +_NOTIONAL_TOLERANCE = 0.01 + + +def _normalized_symbol(value: object) -> str: + symbol = str(value or "").strip().upper() + return symbol.split(".", 1)[0].strip() + + +def _normalized_symbols(value: object) -> set[str]: + if not isinstance(value, (list, tuple, set)): + return set() + return {_normalized_symbol(item) for item in value if _normalized_symbol(item)} + + +def _as_finite_number(value: object, *, field_name: str) -> float: + try: + number = float(value) + except (TypeError, ValueError) as exc: + raise ValueError(f"{field_name} must be numeric") from exc + if not math.isfinite(number): + raise ValueError(f"{field_name} must be finite") + return number + + +def _command_release_findings( + command: ExecutionCommand, + *, + expected_strategy_release: StrategyReleaseIdentity, +) -> tuple[str, ...]: + raw_release = command.intent.get("strategy_release") + if not isinstance(raw_release, Mapping): + return ("release_identity_mismatch",) + try: + command_release = build_strategy_release_identity(raw_release) + except ValueError: + return ("release_identity_invalid",) + if command_release != expected_strategy_release: + return ("release_identity_mismatch",) + return () + + +def _runtime_release_preflight_reason( + receipt: Mapping[str, Any] | None, + *, + expected_strategy_release: StrategyReleaseIdentity, +) -> str | None: + """Refuse to claim commands until the runtime has self-attested its release.""" + if not isinstance(receipt, Mapping): + return "release_receipt_missing" + if str(receipt.get("attestation_state") or "") != "self_attested": + return "release_receipt_missing" + raw_release = receipt.get("strategy_release") + if not isinstance(raw_release, Mapping): + return "release_receipt_missing" + try: + actual_release = build_strategy_release_identity(raw_release) + except ValueError: + return "release_identity_invalid" + if str(receipt.get("release_id") or "") != actual_release.release_id: + return "release_identity_invalid" + if actual_release != expected_strategy_release: + return "release_identity_mismatch" + return None + + +def _build_reconciled_order_proposals( + command: ExecutionCommand, + *, + portfolio: Any, + market_data_port: Any, +) -> tuple[tuple[dict[str, object], ...], tuple[str, ...]]: + """Build dry-run proposals from the immutable target and current holdings. + + Exposure classification compares absolute before/after values for each + position. It is therefore based on reconciled state, never merely on an + order's buy/sell label. + """ + intent = command.intent + if str(intent.get("schema_version") or "") != PAPER_EXECUTION_INTENT_SCHEMA_VERSION: + return (), ("durable_event_history_invalid",) + if str(intent.get("target_mode") or "") != "value": + return (), ("durable_event_history_invalid",) + raw_targets = intent.get("targets") + if not isinstance(raw_targets, Mapping): + return (), ("durable_event_history_invalid",) + + try: + targets = { + _normalized_symbol(symbol): _as_finite_number(target, field_name=f"targets[{symbol!r}]") + for symbol, target in raw_targets.items() + if _normalized_symbol(symbol) + } + except ValueError: + return (), ("durable_event_history_invalid",) + if any(target < 0.0 for target in targets.values()): + # LongBridge's current cash-only value-target contract has no safe + # representation for a short target in this paper consumer. + return (), ("durable_event_history_invalid",) + + strategy_symbols = _normalized_symbols(intent.get("strategy_symbols")) + if not strategy_symbols or not set(targets).issubset(strategy_symbols): + return (), ("durable_event_history_invalid",) + + position_values: dict[str, float] = {} + position_quantities: dict[str, float] = {} + findings: list[str] = [] + for position in tuple(getattr(portfolio, "positions", ()) or ()): + symbol = _normalized_symbol(getattr(position, "symbol", "")) + if not symbol: + findings.append("position_reconciliation_mismatch") + continue + if symbol not in strategy_symbols: + findings.append("position_reconciliation_mismatch") + continue + try: + market_value = _as_finite_number( + getattr(position, "market_value", None), + field_name=f"position[{symbol}].market_value", + ) + quantity = _as_finite_number( + getattr(position, "quantity", None), + field_name=f"position[{symbol}].quantity", + ) + except ValueError: + findings.append("position_reconciliation_mismatch") + continue + if abs(quantity) > _NOTIONAL_TOLERANCE and abs(market_value) <= _NOTIONAL_TOLERANCE: + findings.append("position_reconciliation_mismatch") + position_values[symbol] = position_values.get(symbol, 0.0) + market_value + position_quantities[symbol] = position_quantities.get(symbol, 0.0) + quantity + + cash_balance = getattr(portfolio, "cash_balance", None) + total_equity = getattr(portfolio, "total_equity", None) + if cash_balance is not None and total_equity is not None: + try: + reconciled_total = _as_finite_number(cash_balance, field_name="portfolio.cash_balance") + sum( + position_values.values() + ) + expected_total = _as_finite_number(total_equity, field_name="portfolio.total_equity") + tolerance = max(1.0, abs(expected_total) * 0.005) + if abs(reconciled_total - expected_total) > tolerance: + findings.append("position_reconciliation_mismatch") + except ValueError: + findings.append("position_reconciliation_mismatch") + + proposals: list[dict[str, object]] = [] + for symbol in sorted(strategy_symbols): + current_value = position_values.get(symbol, 0.0) + target_value = targets.get(symbol, 0.0) + delta_value = target_value - current_value + if abs(delta_value) <= _NOTIONAL_TOLERANCE: + continue + try: + quote = market_data_port.get_quote(symbol) + price = _as_finite_number(getattr(quote, "last_price", None), field_name=f"quote[{symbol}].last_price") + if price <= 0.0: + raise ValueError("quote price must be positive") + except Exception: + findings.append("position_reconciliation_mismatch") + continue + + before_exposure = abs(current_value) + after_exposure = abs(target_value) + exposure_delta = after_exposure - before_exposure + if exposure_delta < -_NOTIONAL_TOLERANCE: + exposure_effect = RuntimeCommandExposureEffect.REDUCES + elif exposure_delta > _NOTIONAL_TOLERANCE: + exposure_effect = RuntimeCommandExposureEffect.INCREASES + else: + exposure_effect = RuntimeCommandExposureEffect.NEUTRAL + proposals.append( + { + "symbol": symbol, + "side": "buy" if delta_value > 0.0 else "sell", + "quantity": round(abs(delta_value) / price, 8), + "reference_price": round(price, 8), + "current_value": round(current_value, 8), + "target_value": round(target_value, 8), + "target_notional_delta": round(delta_value, 8), + "current_quantity": round(position_quantities.get(symbol, 0.0), 8), + "exposure_effect": exposure_effect.value, + } + ) + return tuple(proposals), tuple(dict.fromkeys(findings)) + + +def _append_or_raise( + store: ExecutionCommandStore, + command: ExecutionCommand, + *, + next_state: ExecutionCommandState, + expected_previous_state: ExecutionCommandState, + details: Mapping[str, object], +) -> None: + event = store.append_event( + command, + next_state=next_state, + expected_previous_state=expected_previous_state, + details=details, + ) + if event is None: + raise RuntimeError(f"failed to persist paper command event {next_state.value}") + + +def _attempt_reconciliation_required( + store: ExecutionCommandStore, + command: ExecutionCommand, + *, + error: Exception, +) -> None: + try: + state = store.current_state(command) + if state not in { + ExecutionCommandState.CLAIMED, + ExecutionCommandState.SUBMITTED, + ExecutionCommandState.ACCEPTED, + ExecutionCommandState.PARTIALLY_FILLED, + }: + return + store.append_event( + command, + next_state=ExecutionCommandState.RECONCILIATION_REQUIRED, + expected_previous_state=state, + details={ + "paper_simulation": True, + "reason": "consumer_exception_requires_manual_reconciliation", + "error_type": type(error).__name__, + }, + ) + except Exception: + # The original error is already captured by the caller's result. Do + # not risk masking it with a second storage failure. + return + + +def consume_due_paper_execution_commands( + *, + store: ExecutionCommandStore | None, + as_of_session: date | str, + claimant: str, + portfolio: Any, + market_data_port: Any, + runtime_release_receipt: Mapping[str, Any] | None, + expected_strategy_release: StrategyReleaseIdentity | Mapping[str, object] | None, +) -> dict[str, object]: + """Claim and simulate due paper commands; never submit a broker order.""" + if store is None or (not store.cloud_prefix_uri and not store.local_dir): + raise RuntimeError("paper durable execution command store is required") + try: + expected_release = build_strategy_release_identity(expected_strategy_release) + except ValueError: + return { + "schema_version": PAPER_COMMAND_CONSUMER_SCHEMA_VERSION, + "status": "blocked", + "reason": "release_identity_invalid", + "commands": [], + } + release_preflight_reason = _runtime_release_preflight_reason( + runtime_release_receipt, + expected_strategy_release=expected_release, + ) + if release_preflight_reason is not None: + return { + "schema_version": PAPER_COMMAND_CONSUMER_SCHEMA_VERSION, + "status": "blocked", + "reason": release_preflight_reason, + "commands": [], + } + + as_of_date = str(as_of_session)[:10] + commands: list[dict[str, object]] = [] + for command in store.list_due(as_of_date): + if store.current_state(command) is not ExecutionCommandState.QUEUED: + continue + claim = store.claim_due(command, as_of_date=as_of_date, claimant=claimant) + if claim is None: + continue + try: + integrity_findings = list( + _command_release_findings( + command, + expected_strategy_release=expected_release, + ) + ) + if command.execution_mode != "paper": + integrity_findings.append("durable_event_history_invalid") + proposals, reconciliation_findings = _build_reconciled_order_proposals( + command, + portfolio=portfolio, + market_data_port=market_data_port, + ) + integrity_findings.extend(reconciliation_findings) + integrity_findings = list(dict.fromkeys(integrity_findings)) + receipts: list[dict[str, object]] = [] + for proposal in proposals: + decision = evaluate_runtime_command_gate( + action=RuntimeCommandAction.SUBMIT, + exposure_effect=proposal["exposure_effect"], + command=command, + command_state=ExecutionCommandState.CLAIMED, + as_of_session=as_of_date, + runtime_release_receipt=runtime_release_receipt, + expected_strategy_release=expected_release, + integrity_findings=integrity_findings, + policy=RuntimeCommandGatePolicy( + enforcement=RuntimeCommandGateEnforcement.OBSERVE, + ), + ) + receipts.append(decision.to_receipt()) + + # A no-op command still has to pass the command-level release and + # timing checks before it can be closed as paper-filled. + if not proposals: + decision = evaluate_runtime_command_gate( + action=RuntimeCommandAction.SUBMIT, + exposure_effect=RuntimeCommandExposureEffect.NEUTRAL, + command=command, + command_state=ExecutionCommandState.CLAIMED, + as_of_session=as_of_date, + runtime_release_receipt=runtime_release_receipt, + expected_strategy_release=expected_release, + integrity_findings=integrity_findings, + policy=RuntimeCommandGatePolicy( + enforcement=RuntimeCommandGateEnforcement.OBSERVE, + ), + ) + receipts.append(decision.to_receipt()) + + details = { + "paper_simulation": True, + "claimant": claimant, + "integrity_findings": integrity_findings, + "proposals": list(proposals), + "runtime_command_gate_receipts": receipts, + } + if any(not bool(receipt["policy_allows"]) for receipt in receipts): + _append_or_raise( + store, + command, + next_state=ExecutionCommandState.REJECTED, + expected_previous_state=ExecutionCommandState.CLAIMED, + details={ + **details, + "reason": "paper_command_gate_would_block", + }, + ) + commands.append( + { + "command_id": command.command_id, + "status": ExecutionCommandState.REJECTED.value, + "proposals_count": len(proposals), + "would_block": True, + } + ) + continue + + _append_or_raise( + store, + command, + next_state=ExecutionCommandState.SUBMITTED, + expected_previous_state=ExecutionCommandState.CLAIMED, + details=details, + ) + _append_or_raise( + store, + command, + next_state=ExecutionCommandState.ACCEPTED, + expected_previous_state=ExecutionCommandState.SUBMITTED, + details={"paper_simulation": True, "proposals_count": len(proposals)}, + ) + _append_or_raise( + store, + command, + next_state=ExecutionCommandState.FILLED, + expected_previous_state=ExecutionCommandState.ACCEPTED, + details={"paper_simulation": True, "simulated_fill_count": len(proposals)}, + ) + commands.append( + { + "command_id": command.command_id, + "status": ExecutionCommandState.FILLED.value, + "proposals_count": len(proposals), + "would_block": False, + } + ) + except Exception as exc: + _attempt_reconciliation_required(store, command, error=exc) + commands.append( + { + "command_id": command.command_id, + "status": ExecutionCommandState.RECONCILIATION_REQUIRED.value, + "error_type": type(exc).__name__, + } + ) + + return { + "schema_version": PAPER_COMMAND_CONSUMER_SCHEMA_VERSION, + "status": "ok", + "as_of_session": as_of_date, + "commands": commands, + } diff --git a/application/runtime_bootstrap_adapters.py b/application/runtime_bootstrap_adapters.py index e251aba..db8cd7f 100644 --- a/application/runtime_bootstrap_adapters.py +++ b/application/runtime_bootstrap_adapters.py @@ -27,7 +27,8 @@ def _read_app_credentials(self) -> tuple[str, str]: str(self.env_reader(self.app_secret_env_name, "") or ""), ) - def __call__(self) -> tuple[Any, Any, Any]: + def build_contexts(self) -> tuple[Any, Any]: + """Build broker contexts for a normal runtime bootstrap.""" app_key, app_secret = self._read_app_credentials() token = self.refresh_token_if_needed_fn( self.fetch_token_from_secret_fn(self.project_id, self.secret_name), @@ -37,7 +38,16 @@ def __call__(self) -> tuple[Any, Any, Any]: app_secret=app_secret, refresh_threshold_days=self.token_refresh_threshold_days, ) - quote_context, trade_context = self.build_contexts_fn(app_key, app_secret, token) + return self.build_contexts_fn(app_key, app_secret, token) + + def build_read_only_contexts(self) -> tuple[Any, Any]: + """Build broker read contexts without refreshing or mutating a token secret.""" + app_key, app_secret = self._read_app_credentials() + token = self.fetch_token_from_secret_fn(self.project_id, self.secret_name) + return self.build_contexts_fn(app_key, app_secret, token) + + def __call__(self) -> tuple[Any, Any, Any]: + quote_context, trade_context = self.build_contexts() indicators = self.calculate_strategy_indicators_fn(quote_context) if indicators is None: raise Exception("Quote data missing or API limited; cannot compute indicators") diff --git a/application/runtime_composer.py b/application/runtime_composer.py index 5fb6414..b31e91d 100644 --- a/application/runtime_composer.py +++ b/application/runtime_composer.py @@ -207,6 +207,23 @@ def build_rebalance_runtime( fetch_order_status=self.fetch_order_status_fn, ) + def build_read_only_broker_contexts(self) -> tuple[Any, Any]: + """Build LongBridge read contexts without strategy evaluation or an order port.""" + bootstrap = self.bootstrap_builder( + project_id=self.project_id, + secret_name=self.secret_name, + token_refresh_threshold_days=self.token_refresh_threshold_days, + fetch_token_from_secret_fn=self.fetch_token_from_secret_fn, + refresh_token_if_needed_fn=self.refresh_token_if_needed_fn, + build_contexts_fn=self.build_contexts_fn, + calculate_strategy_indicators_fn=lambda _quote_context: {}, + env_reader=self.env_reader, + ) + build_contexts = getattr(bootstrap, "build_read_only_contexts", None) + if not callable(build_contexts): + raise RuntimeError("runtime bootstrap does not support read-only broker contexts") + return build_contexts() + def build_rebalance_config( self, *, diff --git a/docs/paper_execution_command_consumer.md b/docs/paper_execution_command_consumer.md new file mode 100644 index 0000000..9b76bdb --- /dev/null +++ b/docs/paper_execution_command_consumer.md @@ -0,0 +1,39 @@ +# LongBridge 纸面命令消费者 + +这个消费者用于验证延迟执行命令的最后一道风险检查。它只读取 LongBridge 的账户快照和行情，写入纸面命令审计记录；它不会构造执行端口，也不会调用下单 API。 + +## 处理流程 + +1. 策略 dry-run 生成一条不可变的目标仓位命令，并把完整 `strategy_release` 身份写入命令内容。 +2. 专用消费者只认领当日、仍处于 `queued` 的命令。 +3. 它根据实时持仓、现金和行情逐标的生成模拟订单，并比较每个标的前后的绝对风险敞口；不能只根据 `buy` / `sell` 判断风险方向。 +4. 每一笔模拟订单都经过共享的运行时命令门：发布身份、有效交易日、命令事件链、持仓对账任一不一致都会阻止命令完成。 +5. 全部通过时，命令才按 `claimed → submitted → accepted → filled` 记录为纸面模拟完成；任何一笔本应被拦截时，整条命令记录为 `rejected`。消费者异常时转为 `reconciliation_required`，不自动重试或重新下单。 + +命令消费者固定使用 `observe` 模式保存“本应阻止”的回执，但不会因为 observe 模式而模拟一条被拒绝的命令为已成交。 + +## 隔离要求 + +不要使用现有 `longbridge-quant-paper-service`：其当前运行目标并非这条验证链路。改用独立服务，例如 `longbridge-quant-paper-command-verify-service`，并满足以下全部条件： + +- `RUNTIME_TARGET_JSON.execution_mode=paper` 且 `LONGBRIDGE_DRY_RUN_ONLY=true`。 +- `RUNTIME_TARGET_ENABLED=false`，不会创建或恢复定时任务；只能显式调用 `/dry-run` 生成证据和 `/paper-command-consumer` 消费验证。 +- `LONGBRIDGE_EXECUTION_COMMAND_CLOUD_URI` 是新的专用 GCS 前缀，不能等于或位于 `EXECUTION_REPORT_GCS_URI` 之下。 +- `RUNTIME_TARGET_JSON.strategy_release` 必须是完整、已验证的发布身份。身份缺失、无效或与命令不一致时，消费者不认领命令或拒绝该命令。 +- 使用纸面 LongBridge 凭据；即使凭据配置错误地指向其他账户，运行时仍强制 dry-run，消费者也没有下单实现。 + +## 手动部署 + +先准备独立的命令存储 URI 和完整的发布身份 JSON。随后手动发起 workflow；它只作用于新的专用服务，不会改 SG、现有 PAPER 服务或 Cloud Scheduler： + +```bash +gh workflow run sync-cloud-run-env.yml \ + --repo QuantStrategyLab/LongBridgePlatform \ + -f target=paper-command-verify \ + -f paper_command_verify_execution_command_cloud_uri=gs:///longbridge-paper-command-verify \ + -f paper_command_verify_strategy_release_json='' \ + -f deploy_image=true \ + -f sync_env=true +``` + +部署后先手动调用 `/dry-run` 生成命令，再手动调用 `/paper-command-consumer`。检查执行报告中的 `paper_execution_command_consumer`：只有 `status=ok`、命令事件链完整、没有 `would_block` 或 `reconciliation_required`，才可作为后续强制执行阶段的纸面证据。 diff --git a/main.py b/main.py index ea52fe1..823b5b1 100644 --- a/main.py +++ b/main.py @@ -18,6 +18,12 @@ from application.runtime_broker_adapters import build_runtime_broker_adapters from application.runtime_composer import build_runtime_composer from application.rebalance_service import run_strategy as run_rebalance_cycle +from application.durable_execution_commands import ( + resolve_paper_execution_command_consumer_enabled, +) +from application.paper_execution_command_consumer import ( + consume_due_paper_execution_commands, +) from application.runtime_strategy_adapters import build_runtime_strategy_adapters from application.longbridge_execution import submit_order from runtime_execution_policy import fractional_buy_execution_enabled, FRACTIONAL_BUY_QUANTITY_STEP @@ -958,6 +964,106 @@ def run_probe(*, response_body: str = "Probe OK"): print(f"failed to persist execution report: {persist_exc}", flush=True) +def _paper_command_consumer_session_date() -> str: + """Return the exchange-local session date used to claim due commands.""" + try: + import pytz + + return datetime.now(pytz.timezone(MARKET_TIMEZONE)).date().isoformat() + except Exception: + return datetime.now().date().isoformat() + + +def _paper_command_consumer_runtime_is_isolated() -> bool: + runtime_target = getattr(RUNTIME_SETTINGS, "runtime_target", None) + return bool( + getattr(RUNTIME_SETTINGS, "dry_run_only", False) + and runtime_target is not None + and str(getattr(runtime_target, "execution_mode", "") or "").lower() == "paper" + ) + + +def run_paper_execution_command_consumer() -> bool: + """Run an explicitly invoked, read-only paper command consumer. + + This handler never constructs an execution port or calls the LongBridge + order API. It is deliberately separate from ``/run`` so a normal cycle + cannot start consuming delayed commands by accident. + """ + if not _paper_command_consumer_runtime_is_isolated(): + raise RuntimeError( + "paper command consumer requires RUNTIME_TARGET_JSON.execution_mode=paper " + "and LONGBRIDGE_DRY_RUN_ONLY=true" + ) + if not resolve_paper_execution_command_consumer_enabled( + env_reader=os.getenv, + dry_run_only=bool(RUNTIME_SETTINGS.dry_run_only), + ): + raise RuntimeError("paper command consumer is not enabled") + + composer = build_composer() + config = composer.build_rebalance_config() + reporting_adapters = composer.build_reporting_adapters() + log_context, report = reporting_adapters.start_run() + try: + reporting_adapters.log_event( + log_context, + "paper_execution_command_consumer_started", + message="Starting read-only paper execution command consumer", + ) + quote_context, trade_context = composer.build_read_only_broker_contexts() + portfolio = composer.broker_adapters.build_portfolio_port( + quote_context, + trade_context, + ).get_portfolio_snapshot() + result = consume_due_paper_execution_commands( + store=config.execution_command_store, + as_of_session=_paper_command_consumer_session_date(), + claimant=str(os.getenv("K_SERVICE") or "longbridge-paper-command-consumer"), + portfolio=portfolio, + market_data_port=composer.broker_adapters.build_market_data_port(quote_context), + runtime_release_receipt=config.runtime_release_receipt, + expected_strategy_release=config.expected_strategy_release, + ) + report_status = "ok" if result.get("status") == "ok" else "skipped" + finalize_runtime_report( + report, + status=report_status, + summary={"paper_execution_command_consumer": result}, + ) + reporting_adapters.log_event( + log_context, + "paper_execution_command_consumer_completed", + message="Paper execution command consumer completed", + result_status=result.get("status"), + commands_count=len(tuple(result.get("commands") or ())), + ) + return True + except Exception as exc: + append_runtime_report_error( + report, + stage="paper_execution_command_consumer", + message=str(exc), + error_type=type(exc).__name__, + ) + finalize_runtime_report(report, status="error") + reporting_adapters.log_event( + log_context, + "paper_execution_command_consumer_failed", + message="Paper execution command consumer failed", + severity="ERROR", + error_type=type(exc).__name__, + error_message=str(exc), + ) + return False + finally: + try: + report_path = reporting_adapters.persist_execution_report(report) + print(f"execution_report {report_path}", flush=True) + except Exception as persist_exc: + print(f"failed to persist execution report: {persist_exc}", flush=True) + + @app.route("/run", methods=["POST", "GET"]) def handle_trigger(): """Entrypoint for Cloud Run / scheduler: run strategy and return 200.""" @@ -1003,6 +1109,16 @@ def handle_probe(): ) +@app.route("/paper-command-consumer", methods=["POST"]) +def handle_paper_execution_command_consumer(): + """Explicit paper-only command verification endpoint; never scheduler-routed.""" + return _route_with_runtime_error_fallback( + run_paper_execution_command_consumer, + success_body="Paper command consumer OK", + route_label="POST /paper-command-consumer", + ) + + @app.route("/monitor-dispatch", methods=["POST", "GET"]) def handle_monitor_dispatch(): if request_method() == "GET": diff --git a/scripts/build_cloud_run_env_sync_plan.py b/scripts/build_cloud_run_env_sync_plan.py index 85265a9..ef1e4e2 100644 --- a/scripts/build_cloud_run_env_sync_plan.py +++ b/scripts/build_cloud_run_env_sync_plan.py @@ -109,6 +109,7 @@ def _should_add_local_src(candidate: Path) -> bool: "RUNTIME_TARGET_ENABLED", "EXECUTION_REPORT_GCS_URI", "LONGBRIDGE_DURABLE_EXECUTION_COMMAND_PAPER_ENABLED", + "LONGBRIDGE_DURABLE_EXECUTION_COMMAND_PAPER_CONSUMER_ENABLED", "LONGBRIDGE_EXECUTION_COMMAND_CLOUD_URI", ) diff --git a/tests/test_build_cloud_run_env_sync_plan.py b/tests/test_build_cloud_run_env_sync_plan.py index 0e6e8d2..178ee7d 100644 --- a/tests/test_build_cloud_run_env_sync_plan.py +++ b/tests/test_build_cloud_run_env_sync_plan.py @@ -55,6 +55,7 @@ def test_build_cloud_run_env_sync_plan_legacy_mode_uses_shared_env(): "LONGBRIDGE_MARKET_TIMEZONE": "America/New_York", "EXECUTION_REPORT_GCS_URI": "gs://runtime/execution-reports", "LONGBRIDGE_DURABLE_EXECUTION_COMMAND_PAPER_ENABLED": "true", + "LONGBRIDGE_DURABLE_EXECUTION_COMMAND_PAPER_CONSUMER_ENABLED": "true", "LONGBRIDGE_EXECUTION_COMMAND_CLOUD_URI": "gs://runtime/execution-commands/paper", } @@ -77,6 +78,7 @@ def test_build_cloud_run_env_sync_plan_legacy_mode_uses_shared_env(): assert target["env"]["LONGBRIDGE_MARKET"] == "US" assert target["env"]["EXECUTION_REPORT_GCS_URI"] == "gs://runtime/execution-reports" assert target["env"]["LONGBRIDGE_DURABLE_EXECUTION_COMMAND_PAPER_ENABLED"] == "true" + assert target["env"]["LONGBRIDGE_DURABLE_EXECUTION_COMMAND_PAPER_CONSUMER_ENABLED"] == "true" assert target["env"]["LONGBRIDGE_EXECUTION_COMMAND_CLOUD_URI"] == "gs://runtime/execution-commands/paper" assert target["scheduler"] == { "timezone": "America/New_York", diff --git a/tests/test_durable_execution_commands.py b/tests/test_durable_execution_commands.py index 5c038d5..18ea0a2 100644 --- a/tests/test_durable_execution_commands.py +++ b/tests/test_durable_execution_commands.py @@ -11,6 +11,7 @@ from application.durable_execution_commands import ( # noqa: E402 build_paper_execution_command, enqueue_paper_execution_command, + resolve_paper_execution_command_consumer_enabled, resolve_paper_execution_command_producer_enabled, ) from quant_platform_kit.common.strategy_release import build_runtime_loaded_receipt @@ -139,6 +140,19 @@ def enqueue(self, _command): assert result["consumer_authorized"] is False +def test_paper_command_binds_complete_strategy_release_when_available() -> None: + command = build_paper_execution_command( + platform="longbridge", + account_scope="PAPER", + strategy_profile="soxl_soxx_trend_income", + execution=_execution(), + allocation=_allocation(), + strategy_release=_release_identity(), + ) + + assert command.intent["strategy_release"] == _release_identity() + + def test_paper_producer_rejects_live_enablement() -> None: assert resolve_paper_execution_command_producer_enabled( env_reader=lambda _name, _default="": "true", @@ -153,3 +167,19 @@ def test_paper_producer_rejects_live_enablement() -> None: assert "paper-only" in str(exc) else: # pragma: no cover raise AssertionError("live enablement must fail closed") + + +def test_paper_consumer_rejects_live_enablement() -> None: + assert resolve_paper_execution_command_consumer_enabled( + env_reader=lambda _name, _default="": "true", + dry_run_only=True, + ) + try: + resolve_paper_execution_command_consumer_enabled( + env_reader=lambda _name, _default="": "true", + dry_run_only=False, + ) + except RuntimeError as exc: + assert "paper-only" in str(exc) + else: # pragma: no cover + raise AssertionError("live enablement must fail closed") diff --git a/tests/test_paper_execution_command_consumer.py b/tests/test_paper_execution_command_consumer.py new file mode 100644 index 0000000..f6e9c2d --- /dev/null +++ b/tests/test_paper_execution_command_consumer.py @@ -0,0 +1,162 @@ +from __future__ import annotations + +import sys +from datetime import datetime, timezone +from pathlib import Path + + +ROOT = Path(__file__).resolve().parents[1] +if str(ROOT) not in sys.path: + sys.path.insert(0, str(ROOT)) + +from application.durable_execution_commands import build_paper_execution_command # noqa: E402 +from application.paper_execution_command_consumer import ( # noqa: E402 + consume_due_paper_execution_commands, +) +from quant_platform_kit.common.execution_commands import ( # noqa: E402 + ExecutionCommandState, + ExecutionCommandStore, +) +from quant_platform_kit.common.models import ( # noqa: E402 + PortfolioSnapshot, + Position, + QuoteSnapshot, +) +from quant_platform_kit.common.strategy_release import build_runtime_loaded_receipt # noqa: E402 + + +def _release_identity() -> dict[str, str]: + return { + "release_id": "soxl-p2-v3.20260824", + "manifest_sha256": "a" * 64, + "strategy_revision": "soxl-p2-v3", + "config_sha256": "b" * 64, + "risk_policy_sha256": "c" * 64, + "evidence_sha256": "d" * 64, + "plugin_bundle_sha256": "e" * 64, + "effective_session": "2026-08-25", + } + + +def _command(*, bind_release: bool = True): + return build_paper_execution_command( + platform="longbridge", + account_scope="PAPER", + strategy_profile="soxl_soxx_trend_income", + execution={ + "signal_date": "2026-08-24", + "effective_date": "2026-08-25", + "execution_timing_contract": "next_trading_day", + }, + allocation={ + "target_mode": "value", + "targets": {"SOXL": 100.0, "BOXX": 100.0}, + "strategy_symbols": ("SOXL", "BOXX"), + "risk_symbols": ("SOXL",), + "safe_haven_symbols": ("BOXX",), + }, + strategy_release=_release_identity() if bind_release else None, + ) + + +class _MarketDataPort: + def get_quote(self, symbol: str) -> QuoteSnapshot: + return QuoteSnapshot( + symbol=f"{symbol}.US", + as_of=datetime(2026, 8, 25, tzinfo=timezone.utc), + last_price=10.0, + ) + + +def _portfolio(*, include_unmanaged: bool = False) -> PortfolioSnapshot: + positions = [Position(symbol="SOXL", quantity=20.0, market_value=200.0)] + if include_unmanaged: + positions.append(Position(symbol="AAPL", quantity=1.0, market_value=100.0)) + return PortfolioSnapshot( + as_of=datetime(2026, 8, 25, tzinfo=timezone.utc), + total_equity=1000.0 + (100.0 if include_unmanaged else 0.0), + cash_balance=800.0, + positions=tuple(positions), + ) + + +def test_paper_consumer_simulates_reconciled_orders_and_never_calls_an_execution_port(tmp_path: Path) -> None: + store = ExecutionCommandStore(local_dir=tmp_path) + command = _command() + assert store.enqueue(command) + release = _release_identity() + + result = consume_due_paper_execution_commands( + store=store, + as_of_session="2026-08-25", + claimant="paper-command-verify", + portfolio=_portfolio(), + market_data_port=_MarketDataPort(), + runtime_release_receipt=build_runtime_loaded_receipt(strategy_release=release), + expected_strategy_release=release, + ) + + assert result["status"] == "ok" + assert result["commands"] == [ + { + "command_id": command.command_id, + "status": "filled", + "proposals_count": 2, + "would_block": False, + } + ] + assert store.current_state(command) is ExecutionCommandState.FILLED + events = store.events(command) + assert [event.state for event in events] == [ + ExecutionCommandState.CLAIMED, + ExecutionCommandState.SUBMITTED, + ExecutionCommandState.ACCEPTED, + ExecutionCommandState.FILLED, + ] + proposals = events[1].details["proposals"] + assert [proposal["exposure_effect"] for proposal in proposals] == ["increases", "reduces"] + + +def test_paper_consumer_requires_runtime_release_before_claiming(tmp_path: Path) -> None: + store = ExecutionCommandStore(local_dir=tmp_path) + command = _command() + assert store.enqueue(command) + + result = consume_due_paper_execution_commands( + store=store, + as_of_session="2026-08-25", + claimant="paper-command-verify", + portfolio=_portfolio(), + market_data_port=_MarketDataPort(), + runtime_release_receipt=None, + expected_strategy_release=_release_identity(), + ) + + assert result["status"] == "blocked" + assert result["reason"] == "release_receipt_missing" + assert store.current_state(command) is ExecutionCommandState.QUEUED + + +def test_paper_consumer_rejects_unbound_or_unreconciled_commands(tmp_path: Path) -> None: + store = ExecutionCommandStore(local_dir=tmp_path) + command = _command(bind_release=False) + assert store.enqueue(command) + release = _release_identity() + + result = consume_due_paper_execution_commands( + store=store, + as_of_session="2026-08-25", + claimant="paper-command-verify", + portfolio=_portfolio(include_unmanaged=True), + market_data_port=_MarketDataPort(), + runtime_release_receipt=build_runtime_loaded_receipt(strategy_release=release), + expected_strategy_release=release, + ) + + assert result["commands"][0]["status"] == "rejected" + assert result["commands"][0]["would_block"] is True + assert store.current_state(command) is ExecutionCommandState.REJECTED + receipt = store.events(command)[-1].details["runtime_command_gate_receipts"][0] + assert receipt["mode"] == "halted" + assert "release_identity_mismatch" in receipt["reasons"] + assert "position_reconciliation_mismatch" in receipt["reasons"] diff --git a/tests/test_request_handling.py b/tests/test_request_handling.py index cec982f..9f5b1d6 100644 --- a/tests/test_request_handling.py +++ b/tests/test_request_handling.py @@ -252,6 +252,10 @@ def test_cloud_run_route_contracts_are_registered(self): module.app._routes[("/probe", ("POST", "GET"))], module.handle_probe, ) + self.assertIs( + module.app._routes[("/paper-command-consumer", ("POST",))], + module.handle_paper_execution_command_consumer, + ) self.assertIs( module.app._routes[("/monitor-dispatch", ("POST", "GET"))], module.handle_monitor_dispatch, @@ -425,6 +429,88 @@ def fake_run_strategy(*, force_run=False, validation_only=False, validation_labe self.assertTrue(observed["validation_only"]) self.assertEqual(observed["validation_label"], "dry_run") + def test_paper_command_consumer_rejects_any_non_paper_runtime_before_building_composer(self): + module = load_module() + module.build_composer = lambda **_kwargs: (_ for _ in ()).throw( + AssertionError("unsafe runtime must fail before building broker contexts") + ) + + with self.assertRaisesRegex(RuntimeError, "execution_mode=paper"): + module.run_paper_execution_command_consumer() + + def test_paper_command_consumer_uses_read_only_contexts_without_an_execution_port(self): + module = load_module() + observed = {"events": []} + module.RUNTIME_SETTINGS = types.SimpleNamespace( + dry_run_only=True, + runtime_target=build_runtime_target( + platform_id="longbridge", + strategy_profile="soxl_soxx_trend_income", + dry_run_only=True, + deployment_selector="paper-command-verify", + account_scope="paper-command-verify", + service_name="longbridge-quant-paper-command-verify-service", + ), + ) + + class FakePortfolioPort: + def get_portfolio_snapshot(self): + observed["portfolio_read"] = True + return "portfolio-snapshot" + + class FakeBrokerAdapters: + def build_portfolio_port(self, quote_context, trade_context): + observed["contexts"] = (quote_context, trade_context) + return FakePortfolioPort() + + def build_market_data_port(self, quote_context): + observed["market_data_context"] = quote_context + return "market-data-port" + + class FakeComposer: + broker_adapters = FakeBrokerAdapters() + + def build_rebalance_config(self): + return types.SimpleNamespace( + execution_command_store="command-store", + runtime_release_receipt={"attestation_state": "self_attested"}, + expected_strategy_release={"release_id": "release-1"}, + ) + + def build_reporting_adapters(self): + return types.SimpleNamespace( + start_run=lambda: (types.SimpleNamespace(run_id="run-001"), {"status": "pending"}), + log_event=lambda _context, event, **fields: observed["events"].append((event, fields)), + persist_execution_report=lambda report: observed.setdefault("report", dict(report)) or "/tmp/report.json", + ) + + def build_read_only_broker_contexts(self): + observed["read_only_contexts_called"] = True + return "quote-context", "trade-context" + + def fake_consume(**kwargs): + observed["consumer"] = kwargs + return {"status": "ok", "commands": []} + + module.build_composer = lambda **_kwargs: FakeComposer() + module.resolve_paper_execution_command_consumer_enabled = lambda **_kwargs: True + module.consume_due_paper_execution_commands = fake_consume + module.finalize_runtime_report = lambda report, **kwargs: report.update(kwargs) + module._paper_command_consumer_session_date = lambda: "2026-08-25" + + with module.app.test_request_context("/paper-command-consumer", method="POST"): + body, status = module.handle_paper_execution_command_consumer() + + self.assertEqual(status, 200) + self.assertEqual(body, "Paper command consumer OK") + self.assertTrue(observed["read_only_contexts_called"]) + self.assertTrue(observed["portfolio_read"]) + self.assertEqual(observed["contexts"], ("quote-context", "trade-context")) + self.assertEqual(observed["market_data_context"], "quote-context") + self.assertEqual(observed["consumer"]["store"], "command-store") + self.assertEqual(observed["consumer"]["portfolio"], "portfolio-snapshot") + self.assertEqual(observed["consumer"]["market_data_port"], "market-data-port") + def test_handle_probe_checks_account_snapshot_without_success_notification(self): module = load_module() observed = {"override": None, "events": [], "notifications": []} diff --git a/tests/test_runtime_bootstrap_adapters.py b/tests/test_runtime_bootstrap_adapters.py index 2f04d22..5ac3de2 100644 --- a/tests/test_runtime_bootstrap_adapters.py +++ b/tests/test_runtime_bootstrap_adapters.py @@ -73,3 +73,30 @@ def test_build_runtime_bootstrap_raises_when_indicators_unavailable(): assert str(exc) == "Quote data missing or API limited; cannot compute indicators" else: raise AssertionError("expected bootstrap to raise when indicators are unavailable") + + +def test_build_runtime_bootstrap_can_build_read_only_contexts_without_indicators(): + observed = {} + bootstrap = build_runtime_bootstrap( + project_id="project-1", + secret_name="secret-1", + token_refresh_threshold_days=30, + fetch_token_from_secret_fn=lambda *_args, **_kwargs: "refresh-token", + refresh_token_if_needed_fn=lambda *_args, **_kwargs: (_ for _ in ()).throw( + AssertionError("read-only bootstrap must not refresh or mutate a token secret") + ), + build_contexts_fn=lambda app_key, app_secret, token: ( + observed.setdefault("contexts", (app_key, app_secret, token)), + ("quote-context", "trade-context"), + )[-1], + calculate_strategy_indicators_fn=lambda _quote_context: (_ for _ in ()).throw( + AssertionError("read-only bootstrap must not calculate indicators") + ), + env_reader=lambda name, default="": { + "LONGPORT_APP_KEY": "app-key", + "LONGPORT_APP_SECRET": "app-secret", + }.get(name, default), + ) + + assert bootstrap.build_read_only_contexts() == ("quote-context", "trade-context") + assert observed["contexts"] == ("app-key", "app-secret", "refresh-token") diff --git a/tests/test_sync_cloud_run_env_workflow.sh b/tests/test_sync_cloud_run_env_workflow.sh index 127863a..a7f36d8 100644 --- a/tests/test_sync_cloud_run_env_workflow.sh +++ b/tests/test_sync_cloud_run_env_workflow.sh @@ -14,8 +14,15 @@ grep -Fq 'environment: longbridge-sg' "$workflow_file" grep -Fq 'environment: ${{ matrix.target.environment }}' "$workflow_file" grep -Fq 'target:' "$workflow_file" grep -Fq -- '- hk-verify' "$workflow_file" +grep -Fq -- '- paper-command-verify' "$workflow_file" grep -Fq 'INPUT_DEPLOY_IMAGE: ${{ inputs.deploy_image }}' "$workflow_file" grep -Fq 'Apply HK verify-only dispatch defaults' "$workflow_file" +grep -Fq 'Apply isolated paper-command verification defaults' "$workflow_file" +grep -Fq 'paper-command-verify targets only the PAPER deployment' "$workflow_file" +grep -Fq 'paper-command-verify requires a strategy profile, a dedicated command GCS URI, and a complete strategy release identity.' "$workflow_file" +grep -Fq 'paper-command-verify command storage must not reuse the execution report URI or its prefix.' "$workflow_file" +grep -Fq 'echo "RUNTIME_TARGET_ENABLED=false"' "$workflow_file" +grep -Fq 'echo "LONGBRIDGE_DURABLE_EXECUTION_COMMAND_PAPER_CONSUMER_ENABLED=true"' "$workflow_file" grep -Fq 'hk-verify targets only the HK deployment' "$workflow_file" grep -Fq '"strategy_profile": "hk_global_etf_tactical_rotation"' "$workflow_file" grep -Fq 'echo "LONGBRIDGE_DRY_RUN_ONLY=true"' "$workflow_file" @@ -122,6 +129,7 @@ grep -Fq 'IBIT_ZSCORE_EXIT_RISK_OFF_EXPOSURE: ${{ vars.IBIT_ZSCORE_EXIT_RISK_OFF grep -Fq 'IBIT_ZSCORE_EXIT_ALLOW_OUTSIDE_EXECUTION_WINDOW: ${{ vars.IBIT_ZSCORE_EXIT_ALLOW_OUTSIDE_EXECUTION_WINDOW }}' "$workflow_file" grep -Fq 'LONGBRIDGE_DRY_RUN_ONLY: ${{ vars.LONGBRIDGE_DRY_RUN_ONLY }}' "$workflow_file" grep -Fq 'LONGBRIDGE_DURABLE_EXECUTION_COMMAND_PAPER_ENABLED: ${{ vars.LONGBRIDGE_DURABLE_EXECUTION_COMMAND_PAPER_ENABLED }}' "$workflow_file" +grep -Fq 'LONGBRIDGE_DURABLE_EXECUTION_COMMAND_PAPER_CONSUMER_ENABLED: ${{ vars.LONGBRIDGE_DURABLE_EXECUTION_COMMAND_PAPER_CONSUMER_ENABLED }}' "$workflow_file" grep -Fq 'LONGBRIDGE_EXECUTION_COMMAND_CLOUD_URI: ${{ vars.LONGBRIDGE_EXECUTION_COMMAND_CLOUD_URI }}' "$workflow_file" grep -Fq 'RUNTIME_TARGET_JSON: ${{ vars.RUNTIME_TARGET_JSON }}' "$workflow_file" grep -Fq 'ACCOUNT_REGION: ${{ vars.ACCOUNT_REGION || matrix.target.default_account_region }}' "$workflow_file"