fix(risk): envelope target scale + Attention notify (QPK bd8d06e)

application/account_new_risk_gate_support.py

@@ -350,11 +350,122 @@ def new_risk_buy_prohibited(result: NewRiskAdmissionResult) -> bool:
     return result.disposition == NewRiskDisposition.NEW_RISK_PROHIBITED
 
 
-def apply_combined_scale(value: float, scale: float | None) -> float:
-    """Apply a valid reducing scale; missing or out-of-range values are a no-op."""
-    if scale is None or not math.isfinite(scale) or not 0.0 < scale <= 1.0:
-        return value
-    return value * scale
+def apply_combined_scale_to_allocation_targets(
+    allocation: Mapping[str, Any] | None,
+    combined_scale: float | None,
+) -> dict[str, Any]:
+    """Shrink allocation targets by admission combined_scale; omit when scale missing."""
+    from quant_platform_kit.risk.capital_risk_envelope import apply_combined_scale_to_targets
+
+    allocation_out = dict(allocation or {})
+    allocation_out["targets"] = apply_combined_scale_to_targets(
+        allocation_out.get("targets"),
+        combined_scale,
+    )
+    return allocation_out
+
+
+
+_ATTENTION_PLATFORM = "schwab"
+_OPERATIONAL_UNCERTAIN_REASONS = frozenset(
+    {
+        "EQUITY_UNKNOWN_FAIL_CLOSED",
+        "SNAPSHOT_VALIDATION_FAIL_CLOSED",
+        "RECONCILIATION_NOT_VERIFIED",
+        "CIRCUIT_BREAKER_OPEN",
+        "UNKNOWN_PENDING_ORDERS",
+    }
+)
+_attention_sent_keys: set[str] = set()
+
+
+def _resolve_attention_strategy_profile(portfolio: Mapping[str, Any]) -> str:
+    projection = _mapping_or_empty(portfolio.get("account_new_risk_snapshot"))
+    for source in (projection, portfolio, _mapping_or_empty(portfolio.get("metadata"))):
+        value = source.get("strategy_profile")
+        if isinstance(value, str) and value.strip():
+            return value.strip()
+    return str(os.environ.get("STRATEGY_PROFILE") or "").strip() or _DEFAULT_STRATEGY_PROFILE
+
+
+def _resolve_attention_account_alias(
+    portfolio: Mapping[str, Any],
+    execution: Mapping[str, Any] | None,
+) -> str:
+    for source in (
+        portfolio,
+        _mapping_or_empty(portfolio.get("metadata")),
+        _mapping_or_empty(execution),
+    ):
+        for key in ("account_alias", "account_hash", "account_id", "account"):
+            value = source.get(key)
+            if value is not None and str(value).strip():
+                text = str(value).strip()
+                return text[-8:] if len(text) > 8 else text
+    for env_key in ("SCHWAB_ACCOUNT_HASH", "ACCOUNT_ALIAS"):
+        env_alias = str(os.environ.get(env_key) or "").strip()
+        if env_alias:
+            return env_alias[-8:] if len(env_alias) > 8 else env_alias
+    return "unknown"
+
+
+def maybe_publish_attention_for_admission(
+    admission: NewRiskAdmissionResult,
+    *,
+    portfolio: Mapping[str, Any],
+    execution: Mapping[str, Any] | None = None,
+    snapshot: InjectedReconciliationSnapshot | None = None,
+    telegram_sender: Any | None = None,
+    log_message: Any = print,
+) -> Mapping[str, int]:
+    """Publish ACTION/HALT attention when NEW_RISK / ops axes require a page.
+
+    Dedupes on transition keys within the process. Never grants live, raises RRL,
+    or invents daily-loss facts.
+    """
+
+    try:
+        from quant_platform_kit.risk.attention import (
+            AttentionAxes,
+            evaluate_attention,
+            resolve_mandate_dd_budget,
+        )
+        from quant_platform_kit.risk.attention_notify import publish_attention_telegram_transition
+    except ImportError:
+        try:
+            log_message("attention_telegram_skipped reason=attention_api_unavailable")
+        except TypeError:
+            log_message("attention_telegram_skipped reason=attention_api_unavailable", flush=True)
+        return {"sent": 0, "skipped": 1, "failed": 0}
+
+    reasons = tuple(admission.reason_codes or ())
+    prohibited = new_risk_buy_prohibited(admission)
+    operational_uncertain = any(code in _OPERATIONAL_UNCERTAIN_REASONS for code in reasons)
+    profile = _resolve_attention_strategy_profile(portfolio)
+    drawdown = None if snapshot is None else snapshot.drawdown_from_peak
+    decision = evaluate_attention(
+        AttentionAxes(
+            new_risk_prohibited=True if prohibited else None,
+            operational_uncertain=True if operational_uncertain else None,
+            drawdown_from_peak=drawdown,
+            mandate_dd_budget=resolve_mandate_dd_budget(profile),
+        )
+    )
+    return publish_attention_telegram_transition(
+        decision=decision,
+        platform=_ATTENTION_PLATFORM,
+        account_alias=_resolve_attention_account_alias(portfolio, execution),
+        strategy_profile=profile,
+        previous_level=None,
+        already_sent_keys=list(_attention_sent_keys),
+        record_sent_key=_attention_sent_keys.add,
+        telegram_sender=telegram_sender,
+        log_message=log_message,
+    )
+
+
+def reset_attention_sent_keys_for_tests() -> None:
+    _attention_sent_keys.clear()
 
 
 def get_cycle_snapshot() -> InjectedReconciliationSnapshot | None:

application/execution_service.py

@@ -6,12 +6,13 @@
 from dataclasses import dataclass
 
 from application.account_new_risk_gate_support import (
-    apply_combined_scale,
+    apply_combined_scale_to_allocation_targets,
     build_account_new_risk_snapshot,
     build_snapshot_from_portfolio,
     evaluate_cycle_new_risk_admission,
     evaluate_portfolio_new_risk_admission,
     is_account_new_risk_gate_enabled,
+    maybe_publish_attention_for_admission,
     new_risk_buy_prohibited,
     set_cycle_snapshot,
 )
@@ -728,6 +729,31 @@ def buying_power_from_plan(current_portfolio, current_execution):
         # Cloud Logging / dry-run verification need stdout; trade_logs alone are not persisted.
         print(gate_diagnostic_message, flush=True)
         trade_logs.append(gate_diagnostic_message)
+        attention_counts = maybe_publish_attention_for_admission(
+            admission,
+            portfolio=portfolio,
+            execution=execution,
+            snapshot=cycle_snapshot,
+        )
+        attention_message = (
+            "[Attention notify] "
+            f"sent={attention_counts.get('sent', 0)} "
+            f"skipped={attention_counts.get('skipped', 0)} "
+            f"failed={attention_counts.get('failed', 0)}"
+        )
+        print(attention_message, flush=True)
+        trade_logs.append(attention_message)
+        allocation = apply_combined_scale_to_allocation_targets(
+            allocation,
+            admission.combined_scale,
+        )
+        if admission.combined_scale is not None:
+            scale_message = (
+                f"[Envelope scale] combined_scale={admission.combined_scale} "
+                "applied_to_allocation_targets"
+            )
+            print(scale_message, flush=True)
+            trade_logs.append(scale_message)
     else:
         set_cycle_snapshot(None)
 
@@ -797,9 +823,6 @@ def execute_fire_forget(symbol, action_type, quantity, price=None):
             if new_risk_buy_prohibited(admission):
                 record_submitted_order(symbol, action_type, quantity, price, status="rejected")
                 return False
-            quantity = apply_combined_scale(quantity, admission.combined_scale)
-            if action_type != "BUY_NOTIONAL":
-                quantity = int(quantity)
         if action_type == "BUY_NOTIONAL":
             if float(quantity or 0.0) < MIN_NOTIONAL_BUY_USD:
                 return False

pyproject.toml

@@ -19,7 +19,7 @@ dependencies = [
     "google-cloud-storage",
     "google-auth",
     "numpy",
-    "quant-platform-kit @ git+https://github.com/QuantStrategyLab/QuantPlatformKit.git@f982aea79476cadd54d074f7c0447c1111658968",
+    "quant-platform-kit @ git+https://github.com/QuantStrategyLab/QuantPlatformKit.git@bd8d06e50f88927eb525c2c585752435ab24492a",
     "us-equity-strategies @ git+https://github.com/QuantStrategyLab/UsEquityStrategies.git@4a3943883cd6b5bbfe32a559e56a91b40a81b7ce",
 ]
 
@@ -61,5 +61,5 @@ include = [
 
 [tool.uv]
 override-dependencies = [
-  "quant-platform-kit @ git+https://github.com/QuantStrategyLab/QuantPlatformKit.git@f982aea79476cadd54d074f7c0447c1111658968",
+  "quant-platform-kit @ git+https://github.com/QuantStrategyLab/QuantPlatformKit.git@bd8d06e50f88927eb525c2c585752435ab24492a",
 ]

qsl.toml

@@ -5,7 +5,7 @@ upgrade_ring = "ring_d"
 allow_legacy = false
 
 [qsl.requires]
-quant_platform_kit = "f982aea79476cadd54d074f7c0447c1111658968"
+quant_platform_kit = "bd8d06e50f88927eb525c2c585752435ab24492a"
 us_equity_strategies = "4a3943883cd6b5bbfe32a559e56a91b40a81b7ce"
 
 [qsl.compat]

tests/test_account_new_risk_gate.py

@@ -16,11 +16,13 @@
 
 from application.account_new_risk_gate_support import (
     ACCOUNT_NEW_RISK_GATE_ENV,
-    apply_combined_scale,
+    apply_combined_scale_to_allocation_targets,
     build_account_new_risk_snapshot,
     build_snapshot_from_portfolio,
     evaluate_portfolio_new_risk_admission,
+    maybe_publish_attention_for_admission,
     new_risk_buy_prohibited,
+    reset_attention_sent_keys_for_tests,
     set_cycle_snapshot,
 )
 from application.execution_service import execute_rebalance_cycle
@@ -40,6 +42,7 @@
 class AccountNewRiskGateSupportTests(unittest.TestCase):
     def tearDown(self) -> None:
         set_cycle_snapshot(None)
+        reset_attention_sent_keys_for_tests()
         os.environ.pop(ACCOUNT_NEW_RISK_GATE_ENV, None)
         os.environ.pop("SCHWAB_MAX_DAILY_LOSS_USD", None)
         os.environ.pop("MAX_DAILY_LOSS_USD", None)
@@ -348,16 +351,71 @@ def test_rebalance_policy_a_store_read_is_fail_soft_when_unbound(self) -> None:
             result.portfolio["account_new_risk_snapshot"],
         )
 
-    def test_combined_scale_halves_value(self) -> None:
-        self.assertEqual(apply_combined_scale(4.0, 0.5), 2.0)
+    def test_combined_scale_halves_allocation_targets(self) -> None:
+        scaled = apply_combined_scale_to_allocation_targets(
+            {"targets": {"SOXL": 0.6, "SOXX": 0.4}},
+            0.5,
+        )
+        self.assertEqual(scaled["targets"], {"SOXL": 0.3, "SOXX": 0.2})
+
+    def test_missing_combined_scale_leaves_targets(self) -> None:
+        allocation = {"targets": {"SOXL": 0.6}}
+        self.assertEqual(
+            apply_combined_scale_to_allocation_targets(allocation, None)["targets"],
+            {"SOXL": 0.6},
+        )
+
+    def test_attention_notify_on_new_risk_prohibit_dedupes(self) -> None:
+        QPK_ATTENTION = Path(
+            "/Users/lisiyi/Projects/.worktrees/qpk-envelope-scale-20260918/src"
+        )
+        if QPK_ATTENTION.exists() and str(QPK_ATTENTION) not in sys.path:
+            sys.path.insert(0, str(QPK_ATTENTION))
+
+        reset_attention_sent_keys_for_tests()
+        portfolio = {
+            "total_equity": 50_000.0,
+            "strategy_profile": "soxl_soxx_trend_income",
+            "account_hash": "00682abc",
+            "account_new_risk_snapshot": {
+                "production_drift_status": "critical",
+            },
+        }
+        admission = evaluate_portfolio_new_risk_admission(portfolio)
+        self.assertTrue(new_risk_buy_prohibited(admission))
+        snapshot = build_snapshot_from_portfolio(portfolio)
+        payloads: list[str] = []
+
+        def _sender(*, text: str, alert_key: str | None = None, **_kwargs) -> bool:
+            payloads.append(text)
+            return True
+
+        counts = maybe_publish_attention_for_admission(
+            admission,
+            portfolio=portfolio,
+            snapshot=snapshot,
+            telegram_sender=_sender,
+            log_message=lambda *_a, **_k: None,
+        )
+        self.assertEqual(counts.get("sent"), 1)
+        self.assertEqual(len(payloads), 1)
+        counts2 = maybe_publish_attention_for_admission(
+            admission,
+            portfolio=portfolio,
+            snapshot=snapshot,
+            telegram_sender=_sender,
+            log_message=lambda *_a, **_k: None,
+        )
+        self.assertEqual(counts2.get("sent"), 0)
+        self.assertEqual(counts2.get("skipped"), 1)
+        self.assertEqual(len(payloads), 1)
 
-    def test_missing_combined_scale_is_no_op(self) -> None:
-        self.assertEqual(apply_combined_scale(4.0, None), 4.0)
 
 
 class AccountNewRiskGateExecutionCycleTests(unittest.TestCase):
     def tearDown(self) -> None:
         set_cycle_snapshot(None)
+        reset_attention_sent_keys_for_tests()
         os.environ.pop(ACCOUNT_NEW_RISK_GATE_ENV, None)
 
     def _run_buy_cycle(self, *, portfolio_overrides=None):
@@ -481,8 +539,8 @@ def test_execution_cycle_prints_gate_axes_to_stdout(self) -> None:
         self.assertIn("breaker=CLOSED", printed)
         self.assertTrue(any("disposition=ALLOW_NEW_RISK" in log for log in result.trade_logs))
 
-    def test_execution_cycle_halves_buy_quantity_for_half_scale(self) -> None:
-        _result, submitted_orders = self._run_buy_cycle(
+    def test_execution_cycle_scales_targets_for_half_combined_scale(self) -> None:
+        result, submitted_orders = self._run_buy_cycle(
             portfolio_overrides={
                 "total_equity": 40_000.0,
                 "account_new_risk_snapshot": {
@@ -493,7 +551,11 @@ def test_execution_cycle_halves_buy_quantity_for_half_scale(self) -> None:
                 },
             }
         )
+        self.assertTrue(
+            any("applied_to_allocation_targets" in log for log in result.trade_logs)
+        )
         self.assertEqual(len(submitted_orders), 1)
+        # Target value path: half envelope scale → half buy size from full target.
         self.assertEqual(submitted_orders[0].quantity, 2)
 
     def test_execution_cycle_allows_sell_when_buy_prohibited(self) -> None: