feat: attach verified IBKR capital base v2

pyproject.toml

@@ -21,7 +21,7 @@ dependencies = [
     "google-cloud-secret-manager",
     "google-cloud-storage",
     "yfinance",
-    "quant-platform-kit @ git+https://github.com/QuantStrategyLab/QuantPlatformKit.git@2f84b5f1a22b134ea677a94850318ce8251ed40e",
+    "quant-platform-kit @ git+https://github.com/QuantStrategyLab/QuantPlatformKit.git@3c6709019f8ac7db8792ec534f70eaf1d30c5322",
     "us-equity-strategies @ git+https://github.com/QuantStrategyLab/UsEquityStrategies.git@1690b8a84aa819ee6d5063a9fff34bd1c51a5e04",
     "hk-equity-strategies @ git+https://github.com/QuantStrategyLab/HkEquityStrategies.git@8230eb2e3c7681c6c1348c8449773f73d58bfa15",
 ]
@@ -64,5 +64,5 @@ include = [
 
 [tool.uv]
 override-dependencies = [
-  "quant-platform-kit @ git+https://github.com/QuantStrategyLab/QuantPlatformKit.git@2f84b5f1a22b134ea677a94850318ce8251ed40e",
+  "quant-platform-kit @ git+https://github.com/QuantStrategyLab/QuantPlatformKit.git@3c6709019f8ac7db8792ec534f70eaf1d30c5322",
 ]

qsl.toml

@@ -5,7 +5,7 @@ upgrade_ring = "ring_d"
 allow_legacy = false
 
 [qsl.requires]
-quant_platform_kit = "2f84b5f1a22b134ea677a94850318ce8251ed40e"
+quant_platform_kit = "3c6709019f8ac7db8792ec534f70eaf1d30c5322"
 us_equity_strategies = "1690b8a84aa819ee6d5063a9fff34bd1c51a5e04"
 hk_equity_strategies = "8230eb2e3c7681c6c1348c8449773f73d58bfa15"
 

strategy_runtime.py

@@ -12,6 +12,12 @@
     FeatureSnapshotRuntimeSettings,
     evaluate_feature_snapshot_strategy,
 )
+from quant_platform_kit.common.capital_base import (
+    CapitalBaseBinding,
+    CapitalScope,
+    CapitalValuationBasis,
+    build_capital_base_snapshot,
+)
 from quant_platform_kit.common.strategy_plugins import attach_strategy_plugin_metadata
 from quant_platform_kit.ibkr import (
     build_ibkr_strategy_context,
@@ -211,6 +217,74 @@ def _with_consecutive_loss_metadata(self, portfolio_snapshot: Any | None) -> Any
         metadata["consecutive_losses"] = int(streak)
         return replace(portfolio_snapshot, metadata=metadata)
 
+    def _build_capital_base_capabilities(
+        self,
+        portfolio_snapshot: Any | None,
+    ) -> tuple[dict[str, Any], str]:
+        """Attach strict v2 evidence only to an explicitly scoped IBKR snapshot."""
+
+        runtime_target = self.runtime_settings.runtime_target
+        if portfolio_snapshot is None:
+            return {}, "unavailable:portfolio_snapshot"
+        if runtime_target is None:
+            return {}, "unavailable:runtime_target"
+        account_scope = str(runtime_target.account_scope or "").strip()
+        runtime_scope = str(
+            runtime_target.service_name or runtime_target.deployment_selector or ""
+        ).strip()
+        if not account_scope or not runtime_scope:
+            return {}, "unavailable:runtime_scope"
+        metadata = getattr(portfolio_snapshot, "metadata", {})
+        if not isinstance(metadata, Mapping):
+            return {}, "unavailable:portfolio_metadata"
+        if metadata.get("total_equity_source") != "broker_net_liquidation":
+            return {}, "unavailable:unverified_net_liquidation"
+        source_digest = str(metadata.get("source_digest_sha256") or "").strip()
+        if not source_digest:
+            return {}, "unavailable:source_digest"
+        try:
+            capital_base = build_capital_base_snapshot(
+                portfolio_snapshot,
+                account_scope=account_scope,
+                runtime_scope=runtime_scope,
+                strategy_scope=self.profile,
+                reported_currency="USD",
+                target_currency="USD",
+                fx_rate_to_target=1.0,
+                source_digest_sha256=source_digest,
+                capital_scope=CapitalScope.ACCOUNT,
+                valuation_basis=CapitalValuationBasis.BROKER_ACCOUNT_NET_LIQUIDATION,
+            )
+            binding = CapitalBaseBinding(
+                account_scope=account_scope,
+                runtime_scope=runtime_scope,
+                strategy_scope=self.profile,
+                target_currency="USD",
+                capital_scope=CapitalScope.ACCOUNT,
+                valuation_basis=CapitalValuationBasis.BROKER_ACCOUNT_NET_LIQUIDATION,
+            )
+        except (TypeError, ValueError):
+            return {}, "unavailable:invalid_capital_evidence"
+        return {
+            "capital_base": capital_base,
+            "capital_base_binding": binding,
+        }, "verified:broker_account_net_liquidation"
+
+    def _build_context_capabilities(
+        self,
+        *,
+        ib: Any | None,
+        portfolio_snapshot: Any | None,
+    ) -> dict[str, Any]:
+        capabilities: dict[str, Any] = {}
+        if ib is not None:
+            capabilities["broker_client"] = ib
+        capital_base_capabilities, _status = self._build_capital_base_capabilities(
+            portfolio_snapshot
+        )
+        capabilities.update(capital_base_capabilities)
+        return capabilities
+
     def _prepare_portfolio_snapshot(
         self,
         portfolio_snapshot: Any | None,
@@ -624,9 +698,10 @@ def _build_strategy_context(
             market_inputs=market_inputs,
             portfolio_snapshot=portfolio_snapshot,
         )
-        capabilities = {}
-        if ib is not None:
-            capabilities["broker_client"] = ib
+        capabilities = self._build_context_capabilities(
+            ib=ib,
+            portfolio_snapshot=portfolio_snapshot,
+        )
         return build_strategy_context_from_available_inputs(
             entrypoint=self.entrypoint,
             runtime_adapter=context_adapter,
@@ -756,6 +831,7 @@ def _evaluate_direct_market_data_strategy(
         )
         metadata = {
             "strategy_profile": self.profile,
+            "capital_base_status": self._build_capital_base_capabilities(portfolio_snapshot)[1],
             "managed_symbols": managed_symbols,
             "status_icon": self.status_icon,
             "dry_run_only": self.runtime_settings.dry_run_only,
@@ -838,6 +914,7 @@ def _evaluate_market_data_strategy(
         )
         metadata = {
             "strategy_profile": self.profile,
+            "capital_base_status": self._build_capital_base_capabilities(portfolio_snapshot)[1],
             "managed_symbols": managed_symbols,
             "status_icon": self.status_icon,
             "dry_run_only": self.runtime_settings.dry_run_only,
@@ -899,6 +976,13 @@ def _evaluate_value_target_strategy(
             current_holdings=current_holdings,
             ib=ib,
         )
+        ctx = replace(
+            ctx,
+            capabilities=self._build_context_capabilities(
+                ib=ib,
+                portfolio_snapshot=portfolio_snapshot,
+            ),
+        )
         decision = self.entrypoint.evaluate(ctx)
         safe_haven_symbol = next(
             (position.symbol for position in decision.positions if position.role == "safe_haven"),
@@ -916,6 +1000,7 @@ def _evaluate_value_target_strategy(
         metadata = self._enrich_portfolio_metadata(
             {
             "strategy_profile": self.profile,
+            "capital_base_status": self._build_capital_base_capabilities(portfolio_snapshot)[1],
             "managed_symbols": managed_symbols,
             "status_icon": self.status_icon,
             "dry_run_only": self.runtime_settings.dry_run_only,
@@ -1044,9 +1129,10 @@ def build_context(request: FeatureSnapshotContextRequest):
             available_inputs = dict(request.available_inputs)
             if portfolio_snapshot is not None:
                 available_inputs[_PORTFOLIO_SNAPSHOT_INPUT] = portfolio_snapshot
-            capabilities = {}
-            if ib is not None:
-                capabilities["broker_client"] = ib
+            capabilities = self._build_context_capabilities(
+                ib=ib,
+                portfolio_snapshot=portfolio_snapshot,
+            )
             return build_strategy_context_from_available_inputs(
                 entrypoint=request.entrypoint,
                 runtime_adapter=runtime_adapter,

tests/test_strategy_runtime.py

@@ -1,4 +1,5 @@
 import sys
+from dataclasses import replace
 from pathlib import Path
 from types import SimpleNamespace
 
@@ -12,6 +13,7 @@
 
 import strategy_runtime as strategy_runtime_module
 from quant_platform_kit.common.models import PortfolioSnapshot
+from quant_platform_kit.common.runtime_target import build_runtime_target
 from quant_platform_kit.strategy_contracts import (
     PositionTarget,
     StrategyDecision,
@@ -98,6 +100,48 @@ def _build_runtime_settings(
     )
 
 
+def test_loaded_runtime_attaches_v2_capital_base_only_for_verified_ibkr_net_liquidation():
+    entrypoint = SimpleNamespace(manifest=SimpleNamespace(profile="soxl_soxx_trend_income"))
+    runtime = strategy_runtime_module.LoadedStrategyRuntime(
+        entrypoint=entrypoint,
+        runtime_settings=replace(
+            _build_runtime_settings(profile="soxl_soxx_trend_income"),
+            runtime_target=build_runtime_target(
+                platform_id="ibkr",
+                strategy_profile="soxl_soxx_trend_income",
+                dry_run_only=True,
+                account_scope="paper-account-scope",
+                service_name="ibkr-paper-service",
+                execution_environment="dry_run",
+            ),
+        ),
+        runtime_adapter=StrategyRuntimeAdapter(),
+        logger=lambda _message: None,
+    )
+    verified_snapshot = PortfolioSnapshot(
+        as_of=strategy_runtime_module.pd.Timestamp("2026-08-27", tz="UTC").to_pydatetime(),
+        total_equity=1_000.0,
+        metadata={
+            "total_equity_source": "broker_net_liquidation",
+            "source_digest_sha256": "b" * 64,
+        },
+    )
+
+    capabilities, status = runtime._build_capital_base_capabilities(verified_snapshot)
+
+    assert status == "verified:broker_account_net_liquidation"
+    assert capabilities["capital_base"].to_safe_dict()["valuation_basis"] == "broker_account_net_liquidation"
+    assert capabilities["capital_base_binding"].capital_scope.value == "account"
+
+    unverified = replace(
+        verified_snapshot,
+        metadata={"total_equity_source": "unverified_net_liquidation"},
+    )
+    capabilities, status = runtime._build_capital_base_capabilities(unverified)
+    assert capabilities == {}
+    assert status == "unavailable:unverified_net_liquidation"
+
+
 def test_main_compute_signals_uses_strategy_runtime_decision(strategy_module, monkeypatch):
     observed = {}