diff --git a/application/account_new_risk_gate_support.py b/application/account_new_risk_gate_support.py index fb56ac4..63b2bd0 100644 --- a/application/account_new_risk_gate_support.py +++ b/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: diff --git a/application/execution_service.py b/application/execution_service.py index 316bcb6..79b0f2b 100644 --- a/application/execution_service.py +++ b/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 diff --git a/pyproject.toml b/pyproject.toml index aa264a9..b1628d9 100644 --- a/pyproject.toml +++ b/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", ] diff --git a/qsl.toml b/qsl.toml index f7c8cba..83d17c1 100644 --- a/qsl.toml +++ b/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] diff --git a/tests/test_account_new_risk_gate.py b/tests/test_account_new_risk_gate.py index 4cf1d06..70d5499 100644 --- a/tests/test_account_new_risk_gate.py +++ b/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: diff --git a/uv.lock b/uv.lock index 16c4b60..b21726e 100644 --- a/uv.lock +++ b/uv.lock @@ -17,7 +17,7 @@ resolution-markers = [ ] [manifest] -overrides = [{ name = "quant-platform-kit", git = "https://github.com/QuantStrategyLab/QuantPlatformKit.git?rev=f982aea79476cadd54d074f7c0447c1111658968" }] +overrides = [{ name = "quant-platform-kit", git = "https://github.com/QuantStrategyLab/QuantPlatformKit.git?rev=bd8d06e50f88927eb525c2c585752435ab24492a" }] [[package]] name = "anyio" @@ -186,7 +186,7 @@ requires-dist = [ { name = "pytest", marker = "extra == 'test'" }, { name = "pytest-cov", marker = "extra == 'test'" }, { name = "pytz" }, - { name = "quant-platform-kit", git = "https://github.com/QuantStrategyLab/QuantPlatformKit.git?rev=f982aea79476cadd54d074f7c0447c1111658968" }, + { name = "quant-platform-kit", git = "https://github.com/QuantStrategyLab/QuantPlatformKit.git?rev=bd8d06e50f88927eb525c2c585752435ab24492a" }, { name = "requests" }, { name = "ruff", marker = "extra == 'test'" }, { name = "schwab-py" }, @@ -1320,7 +1320,7 @@ wheels = [ [[package]] name = "quant-platform-kit" version = "1.0.0" -source = { git = "https://github.com/QuantStrategyLab/QuantPlatformKit.git?rev=f982aea79476cadd54d074f7c0447c1111658968#f982aea79476cadd54d074f7c0447c1111658968" } +source = { git = "https://github.com/QuantStrategyLab/QuantPlatformKit.git?rev=bd8d06e50f88927eb525c2c585752435ab24492a#bd8d06e50f88927eb525c2c585752435ab24492a" } [[package]] name = "requests"