From 911b2fa9116cf3ff038869d4b3a94417a030ceef Mon Sep 17 00:00:00 2001 From: Pigbibi <20649888+Pigbibi@users.noreply.github.com> Date: Thu, 27 Aug 2026 12:05:39 +0800 Subject: [PATCH 1/2] refactor: share paper command lifecycle Co-Authored-By: Codex --- .../paper_execution_command_consumer.py | 262 +++--------------- docs/paper_execution_command_consumer.md | 2 +- .../test_paper_execution_command_consumer.py | 9 + 3 files changed, 53 insertions(+), 220 deletions(-) diff --git a/application/paper_execution_command_consumer.py b/application/paper_execution_command_consumer.py index e9c2dfe..a2a1645 100644 --- a/application/paper_execution_command_consumer.py +++ b/application/paper_execution_command_consumer.py @@ -16,30 +16,20 @@ from quant_platform_kit.common.execution_commands import ( ExecutionCommand, - ExecutionCommandState, ExecutionCommandStore, - validate_execution_command_release_binding, ) -from quant_platform_kit.common.paper_execution_admission import evaluate_paper_execution_admission +from quant_platform_kit.common.paper_execution_command_consumer import ( + PaperExecutionProposal, + PaperExecutionReconciliation, + consume_due_paper_execution_commands as consume_shared_paper_execution_commands, +) 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, - validate_runtime_loaded_receipt, ) +from quant_platform_kit.common.strategy_release import StrategyReleaseIdentity -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 -_PAPER_COMMAND_GATE_POLICY = RuntimeCommandGatePolicy( - enforcement=RuntimeCommandGateEnforcement.ENFORCE, -) def _normalized_symbol(value: object) -> str: @@ -184,53 +174,34 @@ def _build_reconciled_order_proposals( return tuple(proposals), tuple(dict.fromkeys(findings)) -def _append_or_raise( - store: ExecutionCommandStore, +def _reconcile_command( command: ExecutionCommand, *, - next_state: ExecutionCommandState, - expected_previous_state: ExecutionCommandState, - details: Mapping[str, object], -) -> None: - event = store.append_event( + portfolio: Any, + market_data_port: Any, +) -> PaperExecutionReconciliation: + """Adapt LongBridge's value-target evidence to the shared paper contract.""" + + proposals, integrity_findings = _build_reconciled_order_proposals( command, - next_state=next_state, - expected_previous_state=expected_previous_state, - details=details, + portfolio=portfolio, + market_data_port=market_data_port, + ) + return PaperExecutionReconciliation( + proposals=tuple( + PaperExecutionProposal( + symbol=str(proposal["symbol"]), + exposure_effect=str(proposal["exposure_effect"]), + details={ + key: value + for key, value in proposal.items() + if key not in {"symbol", "exposure_effect"} + }, + ) + for proposal in proposals + ), + integrity_findings=integrity_findings, ) - 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( @@ -243,164 +214,17 @@ def consume_due_paper_execution_commands( 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 = validate_runtime_loaded_receipt( - runtime_release_receipt, - expected_strategy_release=expected_release, - ) - if not release_preflight.is_valid: - return { - "schema_version": PAPER_COMMAND_CONSUMER_SCHEMA_VERSION, - "status": "blocked", - "reason": release_preflight.findings[0], - "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: - admission = evaluate_paper_execution_admission( - command=command, - expected_strategy_release=expected_release, - ) - integrity_findings = list(admission.integrity_findings) - integrity_findings.extend( - validate_execution_command_release_binding( - command, - expected_strategy_release=expected_release, - ).findings - ) - 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=_PAPER_COMMAND_GATE_POLICY, - ) - 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=_PAPER_COMMAND_GATE_POLICY, - ) - receipts.append(decision.to_receipt()) - - details = { - "paper_simulation": True, - "claimant": claimant, - "paper_execution_admission": { - "disposition": admission.disposition.value, - "receipt_sha256": admission.receipt_sha256, - }, - "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__, - } - ) + """Claim and simulate due paper commands through the shared lifecycle.""" - return { - "schema_version": PAPER_COMMAND_CONSUMER_SCHEMA_VERSION, - "status": "ok", - "as_of_session": as_of_date, - "commands": commands, - } + return consume_shared_paper_execution_commands( + store=store, + as_of_session=as_of_session, + claimant=claimant, + reconcile_command=lambda command: _reconcile_command( + command, + portfolio=portfolio, + market_data_port=market_data_port, + ), + runtime_release_receipt=runtime_release_receipt, + expected_strategy_release=expected_strategy_release, + ) diff --git a/docs/paper_execution_command_consumer.md b/docs/paper_execution_command_consumer.md index 69f34b5..45bf771 100644 --- a/docs/paper_execution_command_consumer.md +++ b/docs/paper_execution_command_consumer.md @@ -1,6 +1,6 @@ # LongBridge 纸面命令消费者 -这个消费者用于验证延迟执行命令的最后一道风险检查。它只读取 LongBridge 的账户快照和行情，写入纸面命令审计记录；它不会构造执行端口，也不会调用下单 API。 +这个消费者用于验证延迟执行命令的最后一道风险检查。它只读取 LongBridge 的账户快照和行情，并把平台特有的价值目标转换为共享的纸面提案；命令认领、风险准入、运行时命令门和状态链由 `quant_platform_kit.common.paper_execution_command_consumer` 统一处理。它不会构造执行端口，也不会调用下单 API。 ## 处理流程 diff --git a/tests/test_paper_execution_command_consumer.py b/tests/test_paper_execution_command_consumer.py index 082a0a7..5d2457a 100644 --- a/tests/test_paper_execution_command_consumer.py +++ b/tests/test_paper_execution_command_consumer.py @@ -139,6 +139,15 @@ def test_paper_consumer_simulates_reconciled_orders_and_never_calls_an_execution ] proposals = events[1].details["proposals"] assert [proposal["exposure_effect"] for proposal in proposals] == ["increases", "reduces"] + assert proposals[0]["details"] == { + "side": "buy", + "quantity": 10.0, + "reference_price": 10.0, + "current_value": 0.0, + "target_value": 100.0, + "target_notional_delta": 100.0, + "current_quantity": 0.0, + } receipts = events[1].details["runtime_command_gate_receipts"] assert {receipt["enforcement"] for receipt in receipts} == {"enforce"} assert all(receipt["broker_write_allowed"] is True for receipt in receipts) From 4d88a0ebb3b30d60e6cff2c7fc36032ee8cd48e9 Mon Sep 17 00:00:00 2001 From: Pigbibi <20649888+Pigbibi@users.noreply.github.com> Date: Thu, 27 Aug 2026 12:23:03 +0800 Subject: [PATCH 2/2] fix: bind LongBridge paper commands to runtime scope Co-Authored-By: Codex --- application/paper_execution_command_consumer.py | 2 ++ main.py | 5 +++++ tests/test_paper_execution_command_consumer.py | 12 ++++++++++++ tests/test_request_handling.py | 10 ++++++++++ 4 files changed, 29 insertions(+) diff --git a/application/paper_execution_command_consumer.py b/application/paper_execution_command_consumer.py index a2a1645..c3f3e99 100644 --- a/application/paper_execution_command_consumer.py +++ b/application/paper_execution_command_consumer.py @@ -213,6 +213,7 @@ def consume_due_paper_execution_commands( market_data_port: Any, runtime_release_receipt: Mapping[str, Any] | None, expected_strategy_release: StrategyReleaseIdentity | Mapping[str, object] | None, + expected_command_binding: Mapping[str, object] | None, ) -> dict[str, object]: """Claim and simulate due paper commands through the shared lifecycle.""" @@ -227,4 +228,5 @@ def consume_due_paper_execution_commands( ), runtime_release_receipt=runtime_release_receipt, expected_strategy_release=expected_strategy_release, + expected_command_binding=expected_command_binding, ) diff --git a/main.py b/main.py index 42287a2..33a1fb8 100644 --- a/main.py +++ b/main.py @@ -1027,6 +1027,11 @@ def run_paper_execution_command_consumer() -> bool: 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, + expected_command_binding={ + "platform": "longbridge", + "account_scope": str(config.execution_state_account_scope or "unknown"), + "strategy_profile": str(config.strategy_profile or "unknown"), + }, ) report_status = "ok" if result.get("status") == "ok" else "skipped" finalize_runtime_report( diff --git a/tests/test_paper_execution_command_consumer.py b/tests/test_paper_execution_command_consumer.py index 5d2457a..3f39f45 100644 --- a/tests/test_paper_execution_command_consumer.py +++ b/tests/test_paper_execution_command_consumer.py @@ -104,6 +104,14 @@ def _portfolio(*, include_unmanaged: bool = False) -> PortfolioSnapshot: ) +def _command_binding() -> dict[str, str]: + return { + "platform": "longbridge", + "account_scope": "paper", + "strategy_profile": "soxl_soxx_trend_income", + } + + 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() @@ -118,6 +126,7 @@ def test_paper_consumer_simulates_reconciled_orders_and_never_calls_an_execution market_data_port=_MarketDataPort(), runtime_release_receipt=build_runtime_loaded_receipt(strategy_release=release), expected_strategy_release=release, + expected_command_binding=_command_binding(), ) assert result["status"] == "ok" @@ -169,6 +178,7 @@ def test_paper_consumer_requires_runtime_release_before_claiming(tmp_path: Path) market_data_port=_MarketDataPort(), runtime_release_receipt=None, expected_strategy_release=_release_identity(), + expected_command_binding=_command_binding(), ) assert result["status"] == "blocked" @@ -190,6 +200,7 @@ def test_paper_consumer_rejects_unbound_or_unreconciled_commands(tmp_path: Path) market_data_port=_MarketDataPort(), runtime_release_receipt=build_runtime_loaded_receipt(strategy_release=release), expected_strategy_release=release, + expected_command_binding=_command_binding(), ) assert result["commands"][0]["status"] == "rejected" @@ -217,6 +228,7 @@ def test_paper_consumer_rejects_risk_receipt_bound_to_another_decision(tmp_path: market_data_port=_MarketDataPort(), runtime_release_receipt=build_runtime_loaded_receipt(strategy_release=release), expected_strategy_release=release, + expected_command_binding=_command_binding(), ) assert result["commands"][0]["status"] == "rejected" diff --git a/tests/test_request_handling.py b/tests/test_request_handling.py index 9f5b1d6..be8816e 100644 --- a/tests/test_request_handling.py +++ b/tests/test_request_handling.py @@ -475,6 +475,8 @@ def build_rebalance_config(self): execution_command_store="command-store", runtime_release_receipt={"attestation_state": "self_attested"}, expected_strategy_release={"release_id": "release-1"}, + execution_state_account_scope="paper-command-verify", + strategy_profile=module.STRATEGY_PROFILE, ) def build_reporting_adapters(self): @@ -510,6 +512,14 @@ def fake_consume(**kwargs): self.assertEqual(observed["consumer"]["store"], "command-store") self.assertEqual(observed["consumer"]["portfolio"], "portfolio-snapshot") self.assertEqual(observed["consumer"]["market_data_port"], "market-data-port") + self.assertEqual( + observed["consumer"]["expected_command_binding"], + { + "platform": "longbridge", + "account_scope": "paper-command-verify", + "strategy_profile": module.STRATEGY_PROFILE, + }, + ) def test_handle_probe_checks_account_snapshot_without_success_notification(self): module = load_module()