From 9bb6e3952bbe9ef614943c2eec64061b8544231d Mon Sep 17 00:00:00 2001 From: Pigbibi <20649888+Pigbibi@users.noreply.github.com> Date: Fri, 18 Sep 2026 03:45:25 +0800 Subject: [PATCH] fix(risk): inject Policy A production drift into NEW_RISK gate Read PerformanceStore drift status when not explicit, derive drawdown from peak/equity, and ban new buys on review/critical without optimizing or granting live. Co-authored-by: Cursor --- application/account_new_risk_gate_support.py | 67 ++++++++++++++++---- tests/test_account_new_risk_gate.py | 43 ++++++++++++- 2 files changed, 96 insertions(+), 14 deletions(-) diff --git a/application/account_new_risk_gate_support.py b/application/account_new_risk_gate_support.py index e2c3f69..51f5ea4 100644 --- a/application/account_new_risk_gate_support.py +++ b/application/account_new_risk_gate_support.py @@ -19,9 +19,15 @@ CycleNewRiskHealthEvidence, apply_cycle_new_risk_health_axes, ) +from quant_platform_kit.risk.production_drift_new_risk import ( + resolve_production_drift_status_from_store, +) ACCOUNT_NEW_RISK_GATE_ENV = "ACCOUNT_NEW_RISK_GATE" +_DEFAULT_STRATEGY_PROFILE = "soxl_soxx_trend_income" +_DEFAULT_DOMAIN = "us_equity" + _cycle_snapshot: InjectedReconciliationSnapshot | None = None @@ -119,12 +125,41 @@ def build_account_new_risk_snapshot( return projection def _resolve_production_drift_status(portfolio: Mapping[str, Any], projection: Mapping[str, Any]) -> str | None: - raw = projection.get("production_drift_status") - if raw is None or raw == "": - raw = portfolio.get("production_drift_status") - if raw is None or raw == "": + """Prefer explicit inject; else read-only PerformanceStore (Policy A). Fail-soft.""" + for source in (projection, portfolio, _portfolio_metadata(portfolio)): + raw = source.get("production_drift_status") + if raw is not None and raw != "": + return str(raw).strip() + profile = "" + domain = "" + for source in (projection, portfolio, _portfolio_metadata(portfolio)): + if not profile: + value = source.get("strategy_profile") + if isinstance(value, str) and value.strip(): + profile = value.strip() + if not domain: + value = source.get("strategy_domain") or source.get("domain") + if isinstance(value, str) and value.strip(): + domain = value.strip() + profile = profile or str(os.environ.get("STRATEGY_PROFILE") or "").strip() or _DEFAULT_STRATEGY_PROFILE + domain = domain or str(os.environ.get("STRATEGY_DOMAIN") or "").strip() or _DEFAULT_DOMAIN + return resolve_production_drift_status_from_store( + strategy_profile=profile, + domain=domain, + ) + + +def _resolve_drawdown_from_peak( + *, + equity_usd: float | None, + peak_equity_usd: float | None, + explicit: float | None, +) -> float | None: + if explicit is not None: + return explicit + if equity_usd is None or peak_equity_usd is None or peak_equity_usd <= 0.0: return None - return str(raw).strip() + return max(0.0, 1.0 - (equity_usd / peak_equity_usd)) def build_snapshot_from_portfolio( @@ -137,17 +172,27 @@ def build_snapshot_from_portfolio( equity_usd = _coerce_optional_float(projection.get("equity_usd")) if equity_usd is None: equity_usd = _resolve_equity_usd(portfolio, execution) + peak_equity_usd = ( + _coerce_optional_float(projection.get("peak_equity_usd")) + if "peak_equity_usd" in projection + else _coerce_optional_float(portfolio.get("peak_equity_usd")) + ) + explicit_dd = ( + _coerce_optional_float(projection.get("drawdown_from_peak")) + if "drawdown_from_peak" in projection + else _coerce_optional_float(portfolio.get("drawdown_from_peak")) + ) return InjectedReconciliationSnapshot( observation_status=str(projection.get("observation_status") or "UNAVAILABLE"), reconciliation_status=str(projection.get("reconciliation_status") or "UNVERIFIED"), circuit_breaker_state=str(projection.get("circuit_breaker_state") or "OPEN"), equity_usd=equity_usd, - peak_equity_usd=_coerce_optional_float(projection.get("peak_equity_usd")) - if "peak_equity_usd" in projection - else _coerce_optional_float(portfolio.get("peak_equity_usd")), - drawdown_from_peak=_coerce_optional_float(projection.get("drawdown_from_peak")) - if "drawdown_from_peak" in projection - else _coerce_optional_float(portfolio.get("drawdown_from_peak")), + peak_equity_usd=peak_equity_usd, + drawdown_from_peak=_resolve_drawdown_from_peak( + equity_usd=equity_usd, + peak_equity_usd=peak_equity_usd, + explicit=explicit_dd, + ), realized_vol=_coerce_optional_float(projection.get("realized_vol")) if "realized_vol" in projection else _coerce_optional_float(portfolio.get("realized_vol")), diff --git a/tests/test_account_new_risk_gate.py b/tests/test_account_new_risk_gate.py index c886337..ba67273 100644 --- a/tests/test_account_new_risk_gate.py +++ b/tests/test_account_new_risk_gate.py @@ -69,7 +69,11 @@ def test_healthy_equity_without_explicit_snapshot_allows_new_risk(self) -> None: "equity_usd": 50_000.0, }, ) - result = evaluate_portfolio_new_risk_admission(portfolio) + with patch( + "application.account_new_risk_gate_support.resolve_production_drift_status_from_store", + return_value=None, + ): + result = evaluate_portfolio_new_risk_admission(portfolio) self.assertEqual(result.disposition, NewRiskDisposition.ALLOW_NEW_RISK) def test_snapshot_maps_production_drift_status_from_account_new_risk_snapshot(self) -> None: @@ -110,9 +114,38 @@ def test_invalid_production_drift_status_prohibits_new_risk(self) -> None: def test_absent_production_drift_status_still_allows_when_healthy(self) -> None: portfolio = {"total_strategy_equity": 50_000.0} - result = evaluate_portfolio_new_risk_admission(portfolio) + with patch( + "application.account_new_risk_gate_support.resolve_production_drift_status_from_store", + return_value=None, + ): + result = evaluate_portfolio_new_risk_admission(portfolio) self.assertEqual(result.disposition, NewRiskDisposition.ALLOW_NEW_RISK) + def test_store_critical_production_drift_prohibits_when_status_absent(self) -> None: + portfolio = {"total_strategy_equity": 50_000.0} + with patch( + "application.account_new_risk_gate_support.resolve_production_drift_status_from_store", + return_value="critical", + ) as store_resolver: + result = evaluate_portfolio_new_risk_admission(portfolio) + store_resolver.assert_called_once() + self.assertEqual(result.disposition, NewRiskDisposition.NEW_RISK_PROHIBITED) + self.assertIn("PRODUCTION_DRIFT_CRITICAL", result.reason_codes) + + def test_explicit_production_drift_status_skips_store_lookup(self) -> None: + portfolio = { + "total_strategy_equity": 50_000.0, + "account_new_risk_snapshot": {"production_drift_status": "critical"}, + } + with patch( + "application.account_new_risk_gate_support.resolve_production_drift_status_from_store", + return_value="review", + ) as store_resolver: + result = evaluate_portfolio_new_risk_admission(portfolio) + store_resolver.assert_not_called() + self.assertEqual(result.disposition, NewRiskDisposition.NEW_RISK_PROHIBITED) + self.assertIn("PRODUCTION_DRIFT_CRITICAL", result.reason_codes) + def test_explicit_snapshot_status_wins_over_portfolio_status(self) -> None: portfolio = { "total_strategy_equity": 50_000.0, @@ -177,7 +210,11 @@ def test_drawdown_brake_prohibits_new_risk(self) -> None: def test_equity_without_snapshot_allows_new_risk(self) -> None: """Healthy resolved equity alone now derives cycle-health axes (ALLOW).""" portfolio = {"total_strategy_equity": 50_000.0} - result = evaluate_portfolio_new_risk_admission(portfolio) + with patch( + "application.account_new_risk_gate_support.resolve_production_drift_status_from_store", + return_value=None, + ): + result = evaluate_portfolio_new_risk_admission(portfolio) self.assertEqual(result.disposition, NewRiskDisposition.ALLOW_NEW_RISK) def test_explicit_healthy_snapshot_allows_new_risk(self) -> None: