Merge pull request #373 from QuantStrategyLab/codex/longbridge-account-identity

feat: gate LongBridge orders on account identity

Author: Pigbibi

application/account_identity.py

@@ -0,0 +1,38 @@
+"""Read-only LongBridge account-identity evidence adapter."""
+
+from __future__ import annotations
+
+from typing import Any
+
+from quant_platform_kit.common.account_identity import (
+    AccountIdentityEvidenceSource,
+    BrokerAccountIdentity,
+)
+
+
+def observe_longbridge_account_identity(trade_context: Any) -> BrokerAccountIdentity | None:
+    """Return only the account type exposed by LongBridge's positions API.
+
+    LongBridge's public trade API exposes ``account_channel`` on position
+    channels, but it does not expose a stable account number or paper/live
+    marker for this comparison.  Returning ``None`` on an API failure lets the
+    shared policy produce a redacted evidence-unavailable finding.
+    """
+
+    try:
+        response = trade_context.stock_positions()
+    except Exception:
+        return None
+    channels = getattr(response, "channels", None)
+    if channels is None:
+        return None
+    account_types = tuple(
+        str(getattr(channel, "account_channel", "") or "").strip()
+        for channel in channels
+        if str(getattr(channel, "account_channel", "") or "").strip()
+    )
+    return BrokerAccountIdentity(
+        platform_id="longbridge",
+        evidence_source=AccountIdentityEvidenceSource.BROKER_API_PARTIAL,
+        account_types=account_types,
+    )

application/rebalance_service.py

@@ -9,6 +9,11 @@
 from application.execution_state import build_execution_marker_key
 from application.durable_execution_commands import enqueue_paper_execution_command
 from application.runtime_dependencies import LongBridgeRebalanceConfig, LongBridgeRebalanceRuntime
+from quant_platform_kit.common.account_identity import (
+    AccountIdentityGuardedExecutionPort,
+    AccountIdentityPolicy,
+    evaluate_account_identity,
+)
 from quant_platform_kit.longbridge.market_data import fetch_lot_sizes
 from application.signal_snapshot import build_signal_snapshot
 from notifications.events import NotificationPublisher
@@ -240,6 +245,37 @@ def _durable_command_required_message(*, execution: dict) -> str:
     )
 
 
+def _account_identity_blocked_message(*, findings: tuple[str, ...]) -> str:
+    detail = ", ".join(findings) or "account_identity_blocked"
+    return f"Broker orders blocked by account identity gate: {detail}"
+
+
+def _evaluate_execution_account_identity(
+    *,
+    runtime: LongBridgeRebalanceRuntime,
+    config: LongBridgeRebalanceConfig,
+    trade_context,
+):
+    policy = getattr(config, "account_identity_policy", None)
+    if policy is None:
+        policy = AccountIdentityPolicy()
+    if not isinstance(policy, AccountIdentityPolicy):
+        policy = AccountIdentityPolicy.from_mapping(policy)
+    if not policy.is_configured:
+        return None
+    observer = getattr(runtime, "account_identity_observer", None)
+    observation = observer(trade_context) if callable(observer) else None
+    return evaluate_account_identity(
+        expected_platform_id=getattr(
+            config,
+            "account_identity_expected_platform_id",
+            "longbridge",
+        ),
+        policy=policy,
+        observation=observation,
+    )
+
+
 def _should_record_execution_marker(*, result: ExecutionCycleResult, config: LongBridgeRebalanceConfig) -> bool:
     if not getattr(config, "execution_dedup_enabled", False):
         return False
@@ -293,6 +329,16 @@ def run_strategy(
     quote_context, trade_context, indicators = runtime.bootstrap()
     market_data_port = runtime.market_data_port_factory(quote_context)
     execution_port = runtime.execution_port_factory(trade_context)
+    account_identity_decision = _evaluate_execution_account_identity(
+        runtime=runtime,
+        config=config,
+        trade_context=trade_context,
+    )
+    if account_identity_decision is not None:
+        execution_port = AccountIdentityGuardedExecutionPort(
+            delegate=execution_port,
+            decision=account_identity_decision,
+        )
 
     def load_plan(*, current_snapshot):
         current_snapshot = attach_strategy_plugin_metadata(
@@ -318,6 +364,24 @@ def fetch_replanned_state():
         return load_plan(current_snapshot=current_snapshot)
 
     plan, portfolio, execution, allocation = fetch_replanned_state()
+    account_identity_blocked = bool(
+        account_identity_decision is not None
+        and not account_identity_decision.broker_write_allowed
+    )
+    if account_identity_decision is not None:
+        execution["account_identity"] = account_identity_decision.to_receipt()
+        if account_identity_decision.would_block:
+            print(
+                config.with_prefix(
+                    "account_identity_gate "
+                    f"enforcement={account_identity_decision.policy.enforcement.value} "
+                    f"findings={','.join(account_identity_decision.findings)}"
+                ),
+                flush=True,
+            )
+    if account_identity_blocked:
+        execution["account_identity_blocked"] = True
+        execution["account_identity_block_reason"] = "account_identity_verification_failed"
     paper_command_observation = enqueue_paper_execution_command(
         enabled=bool(getattr(config, "durable_execution_command_paper_enabled", False)),
         dry_run_only=bool(getattr(config, "dry_run_only", False)),
@@ -342,7 +406,7 @@ def fetch_replanned_state():
     if direct_live_routing_blocked:
         execution["direct_live_routing_blocked"] = True
         execution["direct_live_routing_block_reason"] = "durable_execution_command_required"
-    execution_already_recorded = direct_live_routing_blocked
+    execution_already_recorded = direct_live_routing_blocked or account_identity_blocked
     execution_claim_acquired = False
     if not direct_live_routing_blocked and execution_marker_key and execution_state_store:
         try:
@@ -390,7 +454,12 @@ def fetch_replanned_state():
             ) from exc
 
     if execution_already_recorded:
-        if direct_live_routing_blocked:
+        if account_identity_blocked:
+            message = _account_identity_blocked_message(
+                findings=tuple(account_identity_decision.findings),
+            )
+            runtime.notify_issue("Account identity gate blocked broker orders", message)
+        elif direct_live_routing_blocked:
             message = _durable_command_required_message(execution=execution)
             runtime.notify_issue("Next-session execution blocked", message)
         else:

application/runtime_composer.py

@@ -7,6 +7,7 @@
 from typing import Any
 
 from application.runtime_bootstrap_adapters import build_runtime_bootstrap
+from application.account_identity import observe_longbridge_account_identity
 from application.runtime_dependencies import LongBridgeRebalanceConfig, LongBridgeRebalanceRuntime
 from application.execution_state import (
     build_execution_marker_store_from_env,
@@ -22,6 +23,7 @@
 from quant_platform_kit.common.runtime_assembly import build_runtime_assembly
 from quant_platform_kit.common.runtime_target import build_runtime_context_fields
 from quant_platform_kit.common.runtime_target import RuntimeTarget
+from quant_platform_kit.common.account_identity import AccountIdentityPolicy
 from quant_platform_kit.common.strategy_release import build_runtime_loaded_receipt
 from notifications.telegram import build_prefixer
 from quant_platform_kit.notifications.cycle_channel import build_cycle_sender
@@ -205,6 +207,7 @@ def build_rebalance_runtime(
             execution_port_factory=self.broker_adapters.build_execution_port,
             post_submit_order=notification_adapters.post_submit_order,
             fetch_order_status=self.fetch_order_status_fn,
+            account_identity_observer=observe_longbridge_account_identity,
         )
 
     def build_read_only_broker_contexts(self) -> tuple[Any, Any]:
@@ -306,6 +309,12 @@ def build_rebalance_config(
                 if self.runtime_target is not None
                 else None
             ),
+            account_identity_policy=AccountIdentityPolicy.from_mapping(
+                self.runtime_target.account_identity
+                if self.runtime_target is not None
+                else None
+            ),
+            account_identity_expected_platform_id="longbridge",
         )
 
     def load_strategy_plugin_signals(self, raw_mounts):

application/runtime_dependencies.py

@@ -41,6 +41,8 @@ class LongBridgeRebalanceConfig:
     execution_command_store: Any = None
     runtime_release_receipt: Mapping[str, Any] | None = None
     expected_strategy_release: Any = None
+    account_identity_policy: Any = None
+    account_identity_expected_platform_id: str = "longbridge"
 
 
 @dataclass(frozen=True)
@@ -55,3 +57,4 @@ class LongBridgeRebalanceRuntime:
     execution_port_factory: Callable[[Any], ExecutionPort]
     post_submit_order: Callable[[Any, Any, Any], None] | None = None
     fetch_order_status: Callable[..., Any] | None = None
+    account_identity_observer: Callable[[Any], Any] | None = None

main.py

@@ -324,6 +324,9 @@ def _summarize_cycle_result_for_report(cycle_result, *, dry_run: bool) -> dict:
     durable_command = execution.get("durable_execution_command")
     if isinstance(durable_command, dict):
         summary["durable_execution_command"] = dict(durable_command)
+    account_identity = execution.get("account_identity")
+    if isinstance(account_identity, dict):
+        summary["account_identity"] = dict(account_identity)
     return summary
 
 

tests/test_account_identity.py

@@ -0,0 +1,44 @@
+from __future__ import annotations
+
+import sys
+from pathlib import Path
+from types import SimpleNamespace
+
+
+ROOT = Path(__file__).resolve().parents[1]
+if str(ROOT) not in sys.path:
+    sys.path.insert(0, str(ROOT))
+QPK_SRC = ROOT.parent / "QuantPlatformKit" / "src"
+if QPK_SRC.exists() and str(QPK_SRC) not in sys.path:
+    sys.path.insert(0, str(QPK_SRC))
+
+from application.account_identity import observe_longbridge_account_identity
+
+
+def test_longbridge_identity_observer_uses_broker_account_channels_only():
+    observation = observe_longbridge_account_identity(
+        SimpleNamespace(
+            stock_positions=lambda: SimpleNamespace(
+                channels=(
+                    SimpleNamespace(account_channel="Cash"),
+                    SimpleNamespace(account_channel="Margin"),
+                    SimpleNamespace(account_channel="Cash"),
+                )
+            )
+        )
+    )
+
+    assert observation is not None
+    assert observation.platform_id == "longbridge"
+    assert observation.evidence_source.value == "broker_api_partial"
+    assert observation.account_types == ("cash", "margin")
+    assert observation.account_id_fingerprint is None
+    assert observation.account_modes == ()
+
+
+def test_longbridge_identity_observer_returns_no_evidence_when_broker_read_fails():
+    class FailingContext:
+        def stock_positions(self):
+            raise RuntimeError("broker unavailable")
+
+    assert observe_longbridge_account_identity(FailingContext()) is None

tests/test_rebalance_service.py

@@ -25,6 +25,7 @@
     from application import rebalance_service
     from application.execution_service import execute_rebalance_cycle
     from application.runtime_dependencies import LongBridgeRebalanceConfig, LongBridgeRebalanceRuntime
+    from quant_platform_kit.common.account_identity import BrokerAccountIdentity
     from notifications.telegram import build_translator
     from quant_platform_kit.common.models import ExecutionReport, PortfolioSnapshot, Position, QuoteSnapshot
     from quant_platform_kit.common.port_adapters import CallableExecutionPort, CallableMarketDataPort, CallableNotificationPort, CallablePortfolioPort
@@ -156,6 +157,43 @@ def _build_snapshot(plan, *, phase=""):
 
 
 class RebalanceServiceNotificationTests(unittest.TestCase):
+    def test_account_identity_gate_marks_mismatch_for_pre_execution_block(self):
+        runtime = LongBridgeRebalanceRuntime(
+            bootstrap=lambda: None,
+            resolve_rebalance_plan=lambda **_kwargs: {},
+            market_data_port_factory=lambda _value: None,
+            estimate_max_purchase_quantity=lambda *_args, **_kwargs: 0,
+            notifications=CallableNotificationPort(lambda _message: None),
+            notify_issue=lambda _title, _detail: None,
+            portfolio_port_factory=lambda *_args: None,
+            execution_port_factory=lambda _value: None,
+            account_identity_observer=lambda _trade_context: BrokerAccountIdentity(
+                platform_id="longbridge",
+                account_types=("margin",),
+            ),
+        )
+        config = LongBridgeRebalanceConfig(
+            limit_sell_discount=1.0,
+            limit_buy_premium=1.0,
+            separator="-",
+            translator=build_translator("en"),
+            with_prefix=lambda message: message,
+            account_identity_policy={
+                "enforcement": "enforce",
+                "expected_account_types": ["cash"],
+            },
+        )
+
+        decision = rebalance_service._evaluate_execution_account_identity(
+            runtime=runtime,
+            config=config,
+            trade_context=object(),
+        )
+
+        self.assertIsNotNone(decision)
+        self.assertFalse(decision.broker_write_allowed)
+        self.assertIn("account_identity_type_mismatch", decision.findings)
+
     def test_submitted_broker_order_is_recorded_as_pending_reconciliation(self):
         submitted_orders = []
         plan = _build_plan(

tests/test_runtime_composer.py

@@ -142,6 +142,7 @@ def fake_cycle_sender(**kwargs):
     assert runtime.market_data_port_factory == "market-data-port-factory"
     assert runtime.notifications == "notification-port"
     assert runtime.fetch_order_status == "fetch-order-status"
+    assert runtime.account_identity_observer is not None
     silent_runtime.notifications.send_text("precheck heartbeat")
     assert observed["sent_message"] == ("tg-token", "chat-id", "[HK] hello")
     assert runtime.post_submit_order == "post-submit-order"
@@ -155,3 +156,4 @@ def fake_cycle_sender(**kwargs):
     assert config.execution_dedup_enabled is True
     assert config.execution_state_account_scope == "HK"
     assert config.execution_state_store.cloud_prefix_uri == "gs://bucket/runtime-reports"
+    assert config.account_identity_policy.is_configured is False