Skip to content

refactor: share LongBridge paper command lifecycle #387

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
Pigbibi merged 2 commits into from
Aug 27, 2026
Merged

refactor: share LongBridge paper command lifecycle #387

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
911b2fa
refactor: share paper command lifecycle
Pigbibi Aug 27, 2026
4d88a0e
fix: bind LongBridge paper commands to runtime scope
Pigbibi Aug 27, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
264 changes: 45 additions & 219 deletions application/paper_execution_command_consumer.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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:
Expand Down Expand Up @@ -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(
Expand All @@ -242,165 +213,20 @@ 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; 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,
expected_command_binding=expected_command_binding,
)
2 changes: 1 addition & 1 deletion docs/paper_execution_command_consumer.md
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,6 +1,6 @@
# LongBridge 纸面命令消费者

这个消费者用于验证延迟执行命令的最后一道风险检查。它只读取 LongBridge 的账户快照和行情，写入纸面命令审计记录；它不会构造执行端口，也不会调用下单 API。
这个消费者用于验证延迟执行命令的最后一道风险检查。它只读取 LongBridge 的账户快照和行情，并把平台特有的价值目标转换为共享的纸面提案；命令认领、风险准入、运行时命令门和状态链由 `quant_platform_kit.common.paper_execution_command_consumer` 统一处理。它不会构造执行端口，也不会调用下单 API。

## 处理流程

Expand Down
5 changes: 5 additions & 0 deletions main.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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(
Expand Down
21 changes: 21 additions & 0 deletions tests/test_paper_execution_command_consumer.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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()
Expand All @@ -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"
Expand All @@ -139,6 +148,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)
Expand All @@ -160,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"
Expand All @@ -181,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"
Expand Down Expand Up @@ -208,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"
Expand Down
Loading