fix: Recompute account drawdown before risk admission

application/account_new_risk_gate_support.py

@@ -8,6 +8,7 @@
 from collections.abc import Mapping
 from contextlib import contextmanager
 from dataclasses import replace
+from datetime import datetime, timezone
 from typing import Any
 
 from quant_platform_kit.risk.account_new_risk_gate import (
@@ -35,6 +36,11 @@
 # Minimal carrier for gate-only daily-loss axis; not a production RRL binding.
 _DAILY_LOSS_LIMIT_CARRIER_SYMBOL = "SPY"
 
+# A fallback broker measurement is valid for at most one short account-read cycle.
+# Five minutes bounds reuse across delayed/restarted cycles; future dates never qualify.
+BROKER_DRAWDOWN_MAX_AGE_SECONDS = 300
+_DRAWDOWN_TIMESTAMP_KEY = "drawdown_from_peak_observed_at"
+
 _cycle_snapshot: InjectedReconciliationSnapshot | None = None
 
 
@@ -160,13 +166,37 @@ def build_account_new_risk_snapshot(
         if key in portfolio and portfolio[key] != healthy:
             projection[key] = portfolio[key]
     projection["equity_usd"] = equity_usd
-    if "peak_equity_usd" in portfolio and "drawdown_from_peak" not in portfolio:
-        projection.pop("drawdown_from_peak", None)
-    for key in ("peak_equity_usd", "drawdown_from_peak", "realized_vol"):
+    for key in ("peak_equity_usd", "realized_vol"):
         if key in portfolio:
             projection[key] = _risk_input(portfolio[key]) if portfolio[key] is not None else "INVALID_RISK_INPUT"
         elif key in projection:
             projection[key] = _risk_input(projection[key])
+    if "drawdown_from_peak" in portfolio:
+        # A timestamp belongs to its own measurement, never another source's cache.
+        projection.pop(_DRAWDOWN_TIMESTAMP_KEY, None)
+        if _DRAWDOWN_TIMESTAMP_KEY in portfolio:
+            projection[_DRAWDOWN_TIMESTAMP_KEY] = portfolio[_DRAWDOWN_TIMESTAMP_KEY]
+    native_drawdown = _coerce_optional_float(portfolio.get("drawdown_from_peak"))
+    if native_drawdown is not None and native_drawdown >= 0.0:
+        # Keep the current account read's explicit float exactly as main does.
+        # Only strategy/metadata caches need a fresh NAV/peak recomputation.
+        drawdown = native_drawdown
+    elif ("peak_equity_usd" in portfolio and "drawdown_from_peak" not in portfolio
+          and "drawdown_from_peak" not in projection):
+        # Native peak-only reads retain main's absent explicit axis and QPK's
+        # ratio-first calculation; no cached drawdown needs replacement.
+        return projection
+    else:
+        drawdown = _resolve_drawdown_from_peak(
+            equity_usd=equity_usd,
+            peak_equity_usd=_coerce_optional_float(projection.get("peak_equity_usd")),
+            explicit_source=portfolio if "drawdown_from_peak" in portfolio else projection,
+            observed_at=projection.get(_DRAWDOWN_TIMESTAMP_KEY),
+        )
+    projection["drawdown_from_peak"] = drawdown
+    if drawdown is None:
+        # Reuse the existing unavailable-observation gate; None is UNKNOWN, not zero.
+        projection["observation_status"] = "UNAVAILABLE"
     return projection
 
 
@@ -186,10 +216,21 @@ def build_portfolio_from_account_values(
             portfolio[key] = account_values[key]
         elif key in metadata:
             portfolio[key] = metadata[key]
-    for key in ("peak_equity_usd", "drawdown_from_peak", "realized_vol"):
+    for key in ("peak_equity_usd", "drawdown_from_peak", "realized_vol", _DRAWDOWN_TIMESTAMP_KEY):
         if key in account_values:
             portfolio[key] = account_values[key]
+    if "drawdown_from_peak" in account_values and _DRAWDOWN_TIMESTAMP_KEY not in account_values:
+        portfolio.pop(_DRAWDOWN_TIMESTAMP_KEY, None)
     projection = dict(metadata.get("account_new_risk_snapshot") or {})
+    for key in ("peak_equity_usd", "realized_vol"):
+        if key not in account_values and key in metadata:
+            projection[key] = metadata[key]
+    if "drawdown_from_peak" not in account_values and "drawdown_from_peak" in metadata:
+        # Keep late metadata copies in the cache path, including their own timestamp.
+        projection["drawdown_from_peak"] = metadata["drawdown_from_peak"]
+        projection.pop(_DRAWDOWN_TIMESTAMP_KEY, None)
+        if _DRAWDOWN_TIMESTAMP_KEY in metadata:
+            projection[_DRAWDOWN_TIMESTAMP_KEY] = metadata[_DRAWDOWN_TIMESTAMP_KEY]
     # Facts from the native account read outrank a strategy's older projection.
     projection["equity_usd"] = portfolio["total_equity"]
     if "peak_equity_usd" in account_values and "drawdown_from_peak" not in account_values:
@@ -203,9 +244,11 @@ def build_portfolio_from_account_values(
             "account_new_risk_snapshot": projection,
         }
     )
-    for key in ("peak_equity_usd", "drawdown_from_peak", "realized_vol"):
+    for key in ("peak_equity_usd", "realized_vol"):
         if key not in portfolio and key in metadata:
             portfolio[key] = metadata[key]
+    # Derived values stay in the projection so a rebuild cannot mistake a cache
+    # for a native measurement and bypass NAV/peak or timestamp revalidation.
     return portfolio
 
 
@@ -245,13 +288,26 @@ def _resolve_drawdown_from_peak(
     *,
     equity_usd: float | None,
     peak_equity_usd: float | None,
-    explicit: float | None,
+    explicit_source: Mapping[str, Any],
+    observed_at: object = 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:
+    if equity_usd is not None and equity_usd > 0.0 and peak_equity_usd is not None and peak_equity_usd > 0.0:
+        # Difference-first reproduces native 40,000/50,000 -> 0.2 exactly.
+        # Never inspect a cached/broker value when the account read can derive DD.
+        return max(0.0, (peak_equity_usd - equity_usd) / peak_equity_usd)
+    if not isinstance(observed_at, str):
+        return None
+    try:
+        measured_at = datetime.fromisoformat(observed_at.replace("Z", "+00:00"))
+        if measured_at.tzinfo is None:
+            return None
+        age = (datetime.now(timezone.utc) - measured_at).total_seconds()
+    except (ValueError, TypeError, OverflowError):
+        return None
+    if not 0 <= age <= BROKER_DRAWDOWN_MAX_AGE_SECONDS:
         return None
-    return max(0.0, 1.0 - (equity_usd / peak_equity_usd))
+    broker_drawdown = _coerce_optional_float(explicit_source.get("drawdown_from_peak"))
+    return broker_drawdown if broker_drawdown is not None and broker_drawdown >= 0.0 else None
 
 
 def _resolve_explicit_daily_loss_usd(

application/rebalance_service.py

@@ -707,7 +707,7 @@ def _snapshot_to_portfolio_view(snapshot) -> tuple[dict[str, dict[str, float | i
     # Carry explicit snapshot evidence to the existing account new-risk gate.
     # Missing fields stay absent; this projection does not read or reset state.
     for key in ("unknown_pending_orders", "durable_circuit_breaker_state", "peak_equity_usd", "drawdown_from_peak",
-                "realized_vol", "observation_status", "reconciliation_status", "circuit_breaker_state",
+                "drawdown_from_peak_observed_at", "realized_vol", "observation_status", "reconciliation_status", "circuit_breaker_state",
                 "production_drift_status", "daily_loss_usd"):
         if key in metadata:
             account_values[key] = metadata[key]

tests/test_account_new_risk_gate.py

@@ -63,6 +63,7 @@ def test_gate_disabled_when_env_zero():
 
 
 def test_missing_equity_prohibits_new_risk():
+    # Rule A represents unmeasured drawdown as None and retains the existing unavailable-equity prohibition.
     # Missing equity keeps observation_ok=False, which still fails closed via
     # EQUITY_UNKNOWN_FAIL_CLOSED in the gate below -- the circuit-breaker axis
     # is not durably OPEN here since there is no explicit unknown-pending /
@@ -72,6 +73,7 @@ def test_missing_equity_prohibits_new_risk():
         "reconciliation_status": "UNVERIFIED",
         "circuit_breaker_state": "CLOSED",
         "equity_usd": None,
+        "drawdown_from_peak": None,
     }
     result = evaluate_account_values_new_risk_admission({"equity": 0.0})
     assert result.disposition == NewRiskDisposition.NEW_RISK_PROHIBITED
@@ -214,8 +216,10 @@ def test_configured_limit_without_daily_loss_fact_fails_closed():
 
 
 def test_unknown_pending_orders_prohibits_and_opens_breaker():
+    # Rule A requires peak evidence; equal NAV and peak isolate the pending-order OPEN invariant.
     portfolio = {
         "total_equity": 50_000.0,
+        "peak_equity_usd": 50_000.0,
         "unknown_pending_orders": True,
     }
     snapshot = build_account_new_risk_snapshot(portfolio)
@@ -279,6 +283,7 @@ def test_rebalance_snapshot_preserves_absent_and_explicit_risk_fields(metadata):
 
 
 def test_build_portfolio_from_account_values_maps_equity():
+    # Rule A copies peak before projection; (55,000 - 50,000) / 55,000 = 1/11 is independently derived drawdown.
     portfolio = build_portfolio_from_account_values(
         {"equity": 50_000.0},
         signal_metadata={"peak_equity_usd": 55_000.0},
@@ -290,6 +295,8 @@ def test_build_portfolio_from_account_values_maps_equity():
         "reconciliation_status": "VERIFIED",
         "circuit_breaker_state": "CLOSED",
         "equity_usd": 50_000.0,
+        "peak_equity_usd": 55_000.0,
+        "drawdown_from_peak": 1 / 11,
     }
 
 
@@ -448,6 +455,7 @@ def _sender(*, text: str, alert_key: str | None = None, **_kwargs) -> bool:
 
 
 def test_invalid_risk_projection_is_json_safe_and_preserves_independent_brake():
+    # Rule A ignores invalid cached drawdown: measured 0.2 trips the brake while invalid volatility still fails closed.
     import json
     from application.account_new_risk_gate_support import build_account_new_risk_snapshot, evaluate_portfolio_new_risk_admission
     portfolio = {"total_equity": 40000, "peak_equity_usd": 50000, "drawdown_from_peak": float("nan"), "realized_vol": float("nan")}
@@ -457,7 +465,7 @@ def test_invalid_risk_projection_is_json_safe_and_preserves_independent_brake():
     result = evaluate_portfolio_new_risk_admission(portfolio)
     assert result.risk_input_invalid and result.hard_brake_triggered
     assert result.combined_scale == 0.0
-    assert "INVALID_DRAWDOWN_FAIL_CLOSED" in result.reason_codes
+    assert "INVALID_DRAWDOWN_FAIL_CLOSED" not in result.reason_codes
     assert "INVALID_REALIZED_VOL_FAIL_CLOSED" in result.reason_codes
     assert "DRAWDOWN_BRAKE_TRIPPED" in result.reason_codes
 

tests/test_drawdown_cache_postmerge.py

@@ -0,0 +1,195 @@
+"""Post-merge cache regressions; artificial account reads and local journals only."""
+from copy import deepcopy
+from datetime import datetime, timedelta, timezone
+from fractions import Fraction
+import struct
+
+import pytest
+
+from application import account_new_risk_gate_support as gate
+from quant_platform_kit.common.execution_state import AccountSubmissionSession, ExecutionMarkerStore
+
+NOW = datetime(2026, 11, 4, 0, 0, tzinfo=timezone.utc)
+VARIANTS = ["native-snapshot", "top-level-copy"]
+
+
+@pytest.fixture(autouse=True)
+def offline_context(monkeypatch):
+    for key in ("IBKR_MAX_DAILY_LOSS_USD", "MAX_DAILY_LOSS_USD", "RUNTIME_TARGET_JSON"):
+        monkeypatch.delenv(key, raising=False)
+    monkeypatch.setattr(gate, "resolve_production_drift_status_from_store", lambda **_kw: None)
+    monkeypatch.setattr(gate, "is_portfolio_loss_stoploss_enabled", lambda: False)
+
+
+def session_at(path):
+    return AccountSubmissionSession(
+        ExecutionMarkerStore(local_dir=path), broker="fixture", account_id="<PROJECT_ID>",
+        owner_id="<PROJECT_ID>", profile="<PROJECT_ID>", intent_id="1" * 64, release_id="2" * 64,
+    )
+
+
+def facts(variant, equity=40_000, peak=50_000, cached=0.0, timestamp=None):
+    values = {"account_equity_usd": equity}
+    projection = {"observation_status": "COMPLETE", "reconciliation_status": "VERIFIED",
+                  "circuit_breaker_state": "CLOSED"}
+    metadata = {"account_new_risk_snapshot": projection}
+    source = projection if variant == "native-snapshot" else metadata
+    source.update(peak_equity_usd=peak, drawdown_from_peak=cached)
+    if timestamp is not None:
+        source["drawdown_from_peak_observed_at"] = timestamp
+    return values, metadata
+
+
+def projected(values, metadata):
+    return gate.build_snapshot_from_portfolio(
+        gate.build_portfolio_from_account_values(values, signal_metadata=metadata))
+
+
+def bits(number):
+    return struct.pack("!d", number)
+
+
+@pytest.mark.parametrize("variant", VARIANTS)
+@pytest.mark.parametrize("equity,cached", [(40_000, 0.0), (50_000, .6), (50_000, 0.0), (48_000, .6)])
+def test_current_read_wins_and_journal_survives_restart(variant, equity, cached, tmp_path):
+    values, metadata = facts(variant, equity=equity, cached=cached)
+    original = deepcopy((values, metadata))
+    snapshot = projected(values, metadata)
+    # Exact rational arithmetic determines the expected fixture independently.
+    expected = float(max(Fraction(0), Fraction(50_000 - equity, 50_000)))
+    assert bits(snapshot.drawdown_from_peak) == bits(expected)
+    session = session_at(tmp_path)
+    result = gate.evaluate_persistent_account_new_risk_admission(values, signal_metadata=metadata, session=session)
+    breached = equity == 40_000
+    assert gate.new_risk_buy_prohibited(result) is breached
+    assert result.hard_brake_triggered is breached
+    assert session.read_state()["circuit_breaker_open"] is breached
+    # The existing capital envelope retains 85% of targets at equity >= 50,000.
+    expected_target = .425 if equity >= 50_000 else .5
+    assert gate.scale_targets_for_admission({"SPY": .5}, result, snapshot=snapshot) == {"SPY": expected_target}
+    assert (values, metadata) == original
+    if breached:
+        assert "DRAWDOWN_BRAKE_TRIPPED" in result.reason_codes
+        recovered = {**values, "account_equity_usd": 50_000}
+        later = gate.evaluate_persistent_account_new_risk_admission(
+            recovered, signal_metadata=metadata, session=session_at(tmp_path))
+        assert "CIRCUIT_BREAKER_OPEN" in later.reason_codes
+        assert gate.new_risk_buy_prohibited(later)
+
+
+@pytest.mark.parametrize("variant", VARIANTS)
+@pytest.mark.parametrize("missing", [("nav",), ("peak",), ("nav", "peak")])
+def test_missing_measurement_is_unknown_without_cache_fallback(variant, missing, tmp_path):
+    values, metadata = facts(variant, cached=0.0)
+    if "nav" in missing:
+        values.pop("account_equity_usd")
+    if "peak" in missing:
+        metadata.get("account_new_risk_snapshot", {}).pop("peak_equity_usd", None)
+        metadata.pop("peak_equity_usd", None)
+    snapshot = projected(values, metadata)
+    assert snapshot.drawdown_from_peak is None
+    assert snapshot.observation_status == "UNAVAILABLE"
+    session = session_at(tmp_path)
+    result = gate.evaluate_persistent_account_new_risk_admission(values, signal_metadata=metadata, session=session)
+    assert gate.new_risk_buy_prohibited(result)
+    assert "OBSERVATION_NOT_COMPLETE" in result.reason_codes
+    assert not result.hard_brake_triggered
+    assert session.read_state()["circuit_breaker_open"] is False
+    # Unknown measurements block buys without fabricating an exit or a measured breach.
+    assert gate.scale_targets_for_admission({"SPY": .5}, result, snapshot=snapshot) == {"SPY": .5}
+
+
+@pytest.mark.parametrize("variant", VARIANTS)
+@pytest.mark.parametrize("axis", ["nav", "peak"])
+@pytest.mark.parametrize("invalid", [None, True, 0, -1, float("nan"), float("inf"), "invalid"])
+def test_invalid_measurement_is_unknown(variant, axis, invalid):
+    values, metadata = facts(variant, cached=0.0)
+    if axis == "nav":
+        values["account_equity_usd"] = invalid
+    else:
+        source = metadata["account_new_risk_snapshot"] if variant == "native-snapshot" else metadata
+        source["peak_equity_usd"] = invalid
+    snapshot = projected(values, metadata)
+    assert snapshot.drawdown_from_peak is None
+    assert snapshot.observation_status == "UNAVAILABLE"
+    result = gate.evaluate_account_values_new_risk_admission(values, signal_metadata=metadata)
+    assert gate.new_risk_buy_prohibited(result)
+    assert "OBSERVATION_NOT_COMPLETE" in result.reason_codes
+    assert not result.hard_brake_triggered
+
+
+@pytest.mark.parametrize("variant", VARIANTS)
+@pytest.mark.parametrize("stamp", [None, "invalid", "2020-01-01T00:00:00Z", "2999-01-01T00:00:00Z", "2026-11-04T00:00:00"])
+def test_untrusted_broker_timestamp_cannot_authorize_entries(variant, stamp):
+    values, metadata = facts(variant, peak=None, cached=0.0, timestamp=stamp)
+    assert projected(values, metadata).drawdown_from_peak is None
+    assert gate.new_risk_buy_prohibited(gate.evaluate_account_values_new_risk_admission(values, signal_metadata=metadata))
+
+
+@pytest.mark.parametrize("variant", VARIANTS)
+@pytest.mark.parametrize("missing", ["nav", "peak", "both"])
+def test_fresh_broker_value_only_when_recompute_impossible(variant, missing):
+    stamp = datetime.now(timezone.utc).isoformat()
+    values, metadata = facts(variant, cached=.2, timestamp=stamp)
+    source = metadata["account_new_risk_snapshot"] if variant == "native-snapshot" else metadata
+    if missing in {"nav", "both"}:
+        values.pop("account_equity_usd")
+    if missing in {"peak", "both"}:
+        source.pop("peak_equity_usd")
+    assert bits(projected(values, metadata).drawdown_from_peak) == bits(.2)
+    values, metadata = facts(variant, equity=50_000, cached=.2, timestamp=stamp)
+    assert bits(projected(values, metadata).drawdown_from_peak) == bits(0.0)
+    assert not gate.new_risk_buy_prohibited(gate.evaluate_account_values_new_risk_admission(values, signal_metadata=metadata))
+
+
+@pytest.mark.parametrize("variant", VARIANTS)
+def test_timestamp_expires_on_rebuild(variant, monkeypatch):
+    class Clock(datetime):
+        instant = NOW
+
+        @classmethod
+        def now(cls, tz=None):
+            return cls.instant.astimezone(tz)
+
+    monkeypatch.setattr(gate, "datetime", Clock, raising=False)
+    values, metadata = facts(variant, peak=None, cached=0.0, timestamp=NOW.isoformat())
+    portfolio = gate.build_portfolio_from_account_values(values, signal_metadata=metadata)
+    assert gate.build_snapshot_from_portfolio(portfolio).drawdown_from_peak == 0.0
+    Clock.instant = NOW + timedelta(seconds=gate.BROKER_DRAWDOWN_MAX_AGE_SECONDS + 1)
+    assert gate.build_snapshot_from_portfolio(portfolio).drawdown_from_peak is None
+    assert gate.new_risk_buy_prohibited(gate.evaluate_portfolio_new_risk_admission(portfolio))
+
+
+@pytest.mark.parametrize("variant", VARIANTS)
+def test_native_point_two_is_bit_equal(variant):
+    values, metadata = facts(variant, cached=.2)
+    assert bits(projected(values, metadata).drawdown_from_peak) == bits(.2)
+
+
+def test_integer_nav_peak_property_sweep():
+    # Exact integer differences avoid cancellation; the Fraction oracle is independent.
+    for peak in range(1, 201):
+        for nav in range(1, peak + 2):
+            values, metadata = facts("native-snapshot", equity=nav, peak=peak, cached=.6)
+            expected = float(max(Fraction(0), Fraction(peak - nav, peak)))
+            assert bits(projected(values, metadata).drawdown_from_peak) == bits(expected)
+
+
+@pytest.mark.parametrize("variant", VARIANTS)
+def test_recomputation_never_coerces_cached_value(variant):
+    class UnreadableCache:
+        def __float__(self):
+            raise AssertionError("cached value must not be read")
+
+    values, metadata = facts(variant, cached=UnreadableCache())
+    assert bits(projected(values, metadata).drawdown_from_peak) == bits(.2)
+
+
+@pytest.mark.parametrize("variant", VARIANTS)
+def test_fresh_zero_broker_value_allows_without_peak(variant):
+    values, metadata = facts(variant, peak=None, cached=0.0,
+                             timestamp=datetime.now(timezone.utc).isoformat())
+    source = metadata["account_new_risk_snapshot"] if variant == "native-snapshot" else metadata
+    source.pop("peak_equity_usd")
+    result = gate.evaluate_account_values_new_risk_admission(values, signal_metadata=metadata)
+    assert not gate.new_risk_buy_prohibited(result)