From 58af1e1e03623d4abec054b493a9eb05884fb0cb Mon Sep 17 00:00:00 2001 From: Pigbibi <20649888+Pigbibi@users.noreply.github.com> Date: Wed, 4 Nov 2026 09:13:41 +0800 Subject: [PATCH 1/3] fix: Recompute account drawdown before risk admission --- application/account_new_risk_gate_support.py | 59 +++++- tests/test_account_new_risk_gate.py | 10 +- tests/test_drawdown_cache_postmerge.py | 195 +++++++++++++++++++ tests/test_execution_service.py | 63 ++++-- tests/test_pending_order_completeness.py | 8 +- tests/test_persistent_drawdown_breaker.py | 5 +- tests/test_rebalance_service.py | 15 +- tests/test_runtime_broker_adapters.py | 6 +- 8 files changed, 324 insertions(+), 37 deletions(-) create mode 100644 tests/test_drawdown_cache_postmerge.py diff --git a/application/account_new_risk_gate_support.py b/application/account_new_risk_gate_support.py index b01fe6cd..e5125bb3 100644 --- a/application/account_new_risk_gate_support.py +++ b/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 @@ -162,11 +168,26 @@ def build_account_new_risk_snapshot( 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] + 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,9 +207,13 @@ 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] + elif key in metadata: + portfolio[key] = metadata[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 {}) # Facts from the native account read outrank a strategy's older projection. projection["equity_usd"] = portfolio["total_equity"] @@ -203,9 +228,10 @@ def build_portfolio_from_account_values( "account_new_risk_snapshot": projection, } ) - for key in ("peak_equity_usd", "drawdown_from_peak", "realized_vol"): - if key not in portfolio and key in metadata: - portfolio[key] = metadata[key] + # Both native and top-level copy paths expose the same revalidated value. + portfolio["drawdown_from_peak"] = portfolio["account_new_risk_snapshot"]["drawdown_from_peak"] + if _DRAWDOWN_TIMESTAMP_KEY in portfolio["account_new_risk_snapshot"]: + portfolio[_DRAWDOWN_TIMESTAMP_KEY] = portfolio["account_new_risk_snapshot"][_DRAWDOWN_TIMESTAMP_KEY] return portfolio @@ -245,13 +271,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( diff --git a/tests/test_account_new_risk_gate.py b/tests/test_account_new_risk_gate.py index d7ae7810..ff7769da 100644 --- a/tests/test_account_new_risk_gate.py +++ b/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 diff --git a/tests/test_drawdown_cache_postmerge.py b/tests/test_drawdown_cache_postmerge.py new file mode 100644 index 00000000..b587fe53 --- /dev/null +++ b/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="", + owner_id="", profile="", 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) diff --git a/tests/test_execution_service.py b/tests/test_execution_service.py index a251abab..f7c9b2d1 100644 --- a/tests/test_execution_service.py +++ b/tests/test_execution_service.py @@ -311,6 +311,7 @@ def test_execute_rebalance_never_expands_approved_buy_budget( tmp_path, symbol, held, target_value, step, premium, buying_power, reserve, expected_quantity, ): + # Rule A requires measured peak evidence; peak = NAV gives zero drawdown and preserves this order invariant. class FakeIB(SyntheticAccountRefresh): def openTrades(self): return [] @@ -333,7 +334,7 @@ def submit(_ib, intent): _logs, summary = execute_rebalance( FakeIB(), {}, {symbol: {"quantity": held}}, - {"equity": 1000.0, "buying_power": buying_power}, + {"peak_equity_usd": 1000.0, "equity": 1000.0, "buying_power": buying_power}, fetch_quote_snapshots=lambda _ib, symbols: { item: SimpleNamespace(last_price=100.0) for item in symbols }, @@ -429,6 +430,7 @@ def accountValues(self): def test_execute_rebalance_submits_limit_buy_for_underweight_position(monkeypatch, tmp_path): + # Rule A requires measured peak evidence; peak = NAV gives zero drawdown and preserves this order invariant. class FakeIB(SyntheticAccountRefresh): def openTrades(self): return [] @@ -452,7 +454,7 @@ def fake_fetch_quote_snapshots(_ib, symbols): FakeIB(), {"VOO": 1.0}, {}, - {"equity": 1000.0, "buying_power": 1000.0}, + {"peak_equity_usd": 1000.0, "equity": 1000.0, "buying_power": 1000.0}, fetch_quote_snapshots=fake_fetch_quote_snapshots, submit_order_intent=fake_submit_order_intent, order_intent_cls=OrderIntent, @@ -488,6 +490,7 @@ def fake_fetch_quote_snapshots(_ib, symbols): @pytest.mark.parametrize("claim", [None, lambda: False]) def test_execute_rebalance_cannot_submit_without_successful_claim_callback(tmp_path, claim): + # Rule A requires measured peak evidence; peak = NAV gives zero drawdown and preserves this order invariant. submitted = [] ib = with_synthetic_account_refresh(SimpleNamespace( openTrades=lambda: [], reqAllOpenOrders=lambda: [], @@ -495,7 +498,7 @@ def test_execute_rebalance_cannot_submit_without_successful_claim_callback(tmp_p )) with pytest.raises(RuntimeError, match="execution claim required"): execute_rebalance( - ib, {"VOO": 0.5}, {}, {"equity": 1000.0, "buying_power": 1000.0}, + ib, {"VOO": 0.5}, {}, {"peak_equity_usd": 1000.0, "equity": 1000.0, "buying_power": 1000.0}, fetch_quote_snapshots=lambda _ib, symbols: { symbol: SimpleNamespace(last_price=100.0) for symbol in symbols }, @@ -573,6 +576,7 @@ def accountValues(self): def test_execute_rebalance_blocks_when_all_order_submissions_are_rejected(tmp_path): + # Rule A requires measured peak evidence; peak = NAV gives zero drawdown and preserves this order invariant. class FakeIB(SyntheticAccountRefresh): def openTrades(self): return [] @@ -588,7 +592,7 @@ def accountValues(self): FakeIB(), {"VOO": 1.0}, {}, - {"equity": 1000.0, "buying_power": 1000.0}, + {"peak_equity_usd": 1000.0, "equity": 1000.0, "buying_power": 1000.0}, fetch_quote_snapshots=lambda _ib, symbols: { symbol: SimpleNamespace(last_price=100.0) for symbol in symbols }, @@ -624,6 +628,7 @@ def accountValues(self): def test_execute_rebalance_uses_symbol_specific_limit_buy_premium(monkeypatch, tmp_path): + # Rule A requires measured peak evidence; peak = NAV gives zero drawdown and preserves this order invariant. class FakeIB(SyntheticAccountRefresh): def openTrades(self): return [] @@ -644,7 +649,7 @@ def fake_submit_order_intent(_ib, intent): FakeIB(), {"SOXL": 1.0}, {}, - {"equity": 1000.0, "buying_power": 1000.0}, + {"peak_equity_usd": 1000.0, "equity": 1000.0, "buying_power": 1000.0}, fetch_quote_snapshots=lambda _ib, symbols: { symbol: SimpleNamespace(last_price=100.0) for symbol in symbols }, @@ -781,6 +786,7 @@ def accountValues(self): def test_execute_rebalance_hk_profile_dry_run_keeps_whole_share_orders_off_broker(tmp_path): + # Rule A requires measured peak evidence; peak = NAV gives zero drawdown and preserves this order invariant. class FakeIB(SyntheticAccountRefresh): def openTrades(self): return [] @@ -799,7 +805,7 @@ def accountValues(self): FakeIB(), {"02834": 0.45, "03110": 0.35}, {}, - {"equity": 200000.0, "buying_power": 200000.0}, + {"peak_equity_usd": 200000.0, "equity": 200000.0, "buying_power": 200000.0}, fetch_quote_snapshots=lambda _ib, symbols: { symbol: SimpleNamespace(last_price=prices[symbol]) for symbol in symbols }, @@ -876,6 +882,7 @@ def accountValues(self): def test_execute_rebalance_caps_buy_budget_by_reserved_cash_after_sell(tmp_path, monkeypatch): + # Rule A requires measured peak evidence; peak = NAV gives zero drawdown and preserves this order invariant. class FakeIB(SyntheticAccountRefresh): def openTrades(self): return [] @@ -897,7 +904,7 @@ def accountValues(self): "SOXL": {"quantity": 3}, "SOXX": {"quantity": 2}, }, - {"equity": 3289.13, "buying_power": 1412.12}, + {"peak_equity_usd": 3289.13, "equity": 3289.13, "buying_power": 1412.12}, fetch_quote_snapshots=lambda _ib, symbols: { symbol: SimpleNamespace(last_price=prices[symbol]) for symbol in symbols }, @@ -937,6 +944,7 @@ def accountValues(self): def test_execute_rebalance_live_cash_only_uses_initial_settled_cash_when_cash_balance_is_stale( tmp_path, monkeypatch, target_weight, buy_expected, ): + # Rule A requires measured peak evidence; peak = NAV gives zero drawdown and preserves this order invariant. class FakeIB(SyntheticAccountRefresh): def openTrades(self): return [] @@ -973,7 +981,7 @@ def fake_submit_order_intent(_ib, intent): "SOXL": {"quantity": 3}, "SOXX": {"quantity": 2}, }, - {"equity": 1920.36, "buying_power": 112.38}, + {"peak_equity_usd": 1920.36, "equity": 1920.36, "buying_power": 112.38}, fetch_quote_snapshots=lambda _ib, symbols: { symbol: SimpleNamespace(last_price=prices[symbol]) for symbol in symbols }, @@ -1017,6 +1025,7 @@ def fake_submit_order_intent(_ib, intent): def test_execute_rebalance_live_cash_only_does_not_bridge_unfilled_sell_submissions(tmp_path, monkeypatch): + # Rule A requires measured peak evidence; peak = NAV gives zero drawdown and preserves this order invariant. class FakeIB(SyntheticAccountRefresh): def openTrades(self): return [] @@ -1040,7 +1049,7 @@ def fake_submit_order_intent(_ib, intent): FakeIB(), {}, {"SOXL": {"quantity": 3}, "SOXX": {"quantity": 0}}, - {"equity": 700.0, "buying_power": 112.38}, + {"peak_equity_usd": 700.0, "equity": 700.0, "buying_power": 112.38}, fetch_quote_snapshots=lambda _ib, symbols: { symbol: SimpleNamespace(last_price=prices[symbol]) for symbol in symbols }, @@ -1071,6 +1080,7 @@ def fake_submit_order_intent(_ib, intent): def test_execute_rebalance_live_cash_only_uses_partial_fill_proceeds_and_fill_price(tmp_path, monkeypatch): + # Rule A requires measured peak evidence; peak = NAV gives zero drawdown and preserves this order invariant. class FakeIB(SyntheticAccountRefresh): def openTrades(self): return [] @@ -1104,7 +1114,7 @@ def fake_submit_order_intent(_ib, intent): FakeIB(), {}, {"SOXL": {"quantity": 3}, "SOXX": {"quantity": 0}}, - {"equity": 700.0, "buying_power": 100.0}, + {"peak_equity_usd": 700.0, "equity": 700.0, "buying_power": 100.0}, fetch_quote_snapshots=lambda _ib, symbols: { symbol: SimpleNamespace(last_price=prices[symbol]) for symbol in symbols }, @@ -1136,6 +1146,7 @@ def fake_submit_order_intent(_ib, intent): def test_execute_rebalance_projects_unbuyable_weight_target_to_zero_with_settled_funds(tmp_path, monkeypatch): + # Rule A requires measured peak evidence; peak = NAV gives zero drawdown and preserves this order invariant. class FakeIB(SyntheticAccountRefresh): def openTrades(self): return [] @@ -1154,7 +1165,7 @@ def accountValues(self): FakeIB(), {}, {"SOXX": {"quantity": 1}}, - {"equity": 773.69, "buying_power": 236.81}, + {"peak_equity_usd": 773.69, "equity": 773.69, "buying_power": 236.81}, fetch_quote_snapshots=lambda _ib, symbols: { symbol: SimpleNamespace(last_price=prices[symbol]) for symbol in symbols }, @@ -1313,6 +1324,7 @@ def accountValues(self): def test_execute_rebalance_routes_order_to_single_account_id(monkeypatch, tmp_path): + # Rule A requires measured peak evidence; peak = NAV gives zero drawdown and preserves this order invariant. class FakeIB(SyntheticAccountRefresh): def openTrades(self): return [] @@ -1339,7 +1351,7 @@ def fake_submit_order_intent(_ib, intent): FakeIB(), {"TQQQ": 1.0}, {}, - {"equity": 1000.0, "buying_power": 1000.0}, + {"peak_equity_usd": 1000.0, "equity": 1000.0, "buying_power": 1000.0}, fetch_quote_snapshots=lambda _ib, symbols: { symbol: SimpleNamespace(last_price=100.0) for symbol in symbols }, @@ -1505,6 +1517,7 @@ def fake_submit_order_intent(_ib, intent): def test_execute_rebalance_projects_unbuyable_soxx_target_to_cash(monkeypatch, tmp_path): + # Rule A requires measured peak evidence; peak = NAV gives zero drawdown and preserves this order invariant. class FakeIB(SyntheticAccountRefresh): def openTrades(self): return [] @@ -1523,7 +1536,7 @@ def accountValues(self): FakeIB(), {}, {"SOXL": {"quantity": 0}, "SOXX": {"quantity": 1}, "BOXX": {"quantity": 0}}, - {"equity": 1170.60, "buying_power": 519.54}, + {"peak_equity_usd": 1170.60, "equity": 1170.60, "buying_power": 519.54}, fetch_quote_snapshots=lambda _ib, symbols: { symbol: SimpleNamespace(last_price=prices[symbol]) for symbol in symbols }, @@ -1571,6 +1584,7 @@ def accountValues(self): def test_execute_rebalance_caps_buy_to_cash_when_overweight_sell_rounds_to_zero( monkeypatch, tmp_path ): + # Rule A requires measured peak evidence; peak = NAV gives zero drawdown and preserves this order invariant. """Reproduce SOXL/SOXX rotation: SOXX trim rounds to 0 shares; buys must not exceed cash.""" class FakeIB(SyntheticAccountRefresh): def openTrades(self): @@ -1593,7 +1607,7 @@ def accountValues(self): FakeIB(), {}, {"SOXL": {"quantity": 0}, "SOXX": {"quantity": 4}, "BOXX": {"quantity": 0}}, - {"equity": 4032.0, "buying_power": 400.0}, + {"peak_equity_usd": 4032.0, "equity": 4032.0, "buying_power": 400.0}, fetch_quote_snapshots=lambda _ib, symbols: { symbol: SimpleNamespace(last_price=prices[symbol]) for symbol in symbols }, @@ -1901,6 +1915,7 @@ def accountValues(self): def test_execute_rebalance_returns_structured_summary_when_requested(monkeypatch, tmp_path): + # Rule A requires measured peak evidence; peak = NAV gives zero drawdown and preserves this order invariant. class FakeIB(SyntheticAccountRefresh): def openTrades(self): return [] @@ -1921,7 +1936,7 @@ def fake_fetch_quote_snapshots(_ib, symbols): FakeIB(), {"VOO": 0.8, "BOXX": 0.2}, {}, - {"equity": 1000.0, "buying_power": 1000.0}, + {"peak_equity_usd": 1000.0, "equity": 1000.0, "buying_power": 1000.0}, fetch_quote_snapshots=fake_fetch_quote_snapshots, submit_order_intent=lambda *_args, **_kwargs: SimpleNamespace(broker_order_id="1", status="Submitted"), order_intent_cls=OrderIntent, @@ -1963,6 +1978,7 @@ def fake_fetch_quote_snapshots(_ib, symbols): def test_execute_rebalance_keeps_small_safe_haven_target_as_cash(monkeypatch, tmp_path): + # Rule A requires measured peak evidence; peak = NAV gives zero drawdown and preserves this order invariant. class FakeIB(SyntheticAccountRefresh): def openTrades(self): return [] @@ -1980,7 +1996,7 @@ def accountValues(self): FakeIB(), {"VOO": 0.5, "BOXX": 0.5}, {}, - {"equity": 1500.0, "buying_power": 1500.0}, + {"peak_equity_usd": 1500.0, "equity": 1500.0, "buying_power": 1500.0}, fetch_quote_snapshots=lambda _ib, symbols: { symbol: SimpleNamespace(last_price=100.0) for symbol in symbols }, @@ -2015,6 +2031,7 @@ def accountValues(self): def test_execute_rebalance_sells_cash_sweep_symbol_when_buying_power_is_short(monkeypatch, tmp_path): + # Rule A requires measured peak evidence; peak = NAV gives zero drawdown and preserves this order invariant. class FakeIB(SyntheticAccountRefresh): def __init__(self): self._account_values_calls = 0 @@ -2055,7 +2072,7 @@ def fake_submit_order_intent(_ib, intent): FakeIB(), {"VOO": 0.8, "BOXX": 0.2}, {"VOO": {"quantity": 0}, "BOXX": {"quantity": 2}}, - {"equity": 1000.0, "buying_power": 24.0}, + {"peak_equity_usd": 1000.0, "equity": 1000.0, "buying_power": 24.0}, fetch_quote_snapshots=lambda *_args, **_kwargs: { "VOO": SimpleNamespace(last_price=100.0), "BOXX": SimpleNamespace(last_price=100.0), @@ -2185,6 +2202,7 @@ def accountValues(self): def test_execute_rebalance_uses_snapshot_prices_for_dry_run_when_quotes_missing(tmp_path): + # Rule A requires measured peak evidence; peak = NAV gives zero drawdown and preserves this order invariant. class FakeIB(SyntheticAccountRefresh): def openTrades(self): return [] @@ -2200,7 +2218,7 @@ def accountValues(self): FakeIB(), {"VOO": 0.6, "BOXX": 0.4}, {}, - {"equity": 3000.0, "buying_power": 3000.0}, + {"peak_equity_usd": 3000.0, "equity": 3000.0, "buying_power": 3000.0}, fetch_quote_snapshots=lambda *_args, **_kwargs: {}, submit_order_intent=lambda *_args, **_kwargs: None, order_intent_cls=OrderIntent, @@ -2236,6 +2254,7 @@ def accountValues(self): def test_execute_rebalance_uses_price_fallbacks_for_live_when_quotes_missing(monkeypatch, tmp_path): + # Rule A requires measured peak evidence; peak = NAV gives zero drawdown and preserves this order invariant. class FakeIB(SyntheticAccountRefresh): def openTrades(self): return [] @@ -2259,7 +2278,7 @@ def fake_submit_order_intent(_ib, intent): FakeIB(), {"VOO": 1.0}, {}, - {"equity": 1000.0, "buying_power": 1000.0}, + {"peak_equity_usd": 1000.0, "equity": 1000.0, "buying_power": 1000.0}, fetch_quote_snapshots=lambda *_args, **_kwargs: {}, submit_order_intent=fake_submit_order_intent, order_intent_cls=OrderIntent, @@ -2295,6 +2314,7 @@ def fake_submit_order_intent(_ib, intent): @pytest.fixture def uncertain_cycle(tmp_path, monkeypatch): + # Rule A requires an explicit fake peak equal to NAV so the original order checks exercise a measured zero drawdown. """Real executor/claim/identity lock with offline approval and broker reports.""" from application import execution_service as service @@ -2367,7 +2387,7 @@ def submit(_ib, intent): ) state.session = session return _execute_rebalance( - FakeIB(), targets, positions, {"equity": 2000.0, "buying_power": 1500.0}, + FakeIB(), targets, positions, {"peak_equity_usd": 2000.0, "equity": 2000.0, "buying_power": 1500.0}, fetch_quote_snapshots=lambda _ib, syms: { s: SimpleNamespace(last_price=50.0) for s in syms }, @@ -2435,6 +2455,7 @@ def test_missing_broker_report_stops_without_discarding_intent(uncertain_cycle, def test_unknown_buy_stops_later_buy(uncertain_cycle): + # Rule A requires measured peak evidence; peak = NAV gives zero drawdown and preserves this order invariant. _logs, summary = uncertain_cycle.run( SimpleNamespace(broker_order_id="buy-unknown", status="Unknown"), first_side="buy", ) @@ -2482,6 +2503,7 @@ def test_claim_persistence_failure_prevents_submit(uncertain_cycle): @pytest.mark.parametrize("status", ["Rejected", "rejected", "Cancelled", "ApiCancelled", "Submitted"]) def test_known_outcome_retains_existing_continuation(uncertain_cycle, status): + # Rule A requires measured peak evidence; peak = NAV gives zero drawdown and preserves this order invariant. _logs, summary = uncertain_cycle.run(SimpleNamespace(broker_order_id="known-order", status=status)) assert [(i.symbol, i.side) for i in uncertain_cycle.submitted] == [ ("TQQQ", "sell"), ("QQQM", "buy"), @@ -2492,6 +2514,7 @@ def test_known_outcome_retains_existing_continuation(uncertain_cycle, status): @pytest.mark.parametrize("failure", ["disconnect", "receipt-persistence"]) def test_partial_fill_then_failure_retains_actual_adapter_facts(uncertain_cycle, failure): + # Rule A requires measured peak evidence; peak = NAV gives zero drawdown and preserves this order invariant. from application.ibkr_order_execution import submit_order_intent class DisconnectedStatus: diff --git a/tests/test_pending_order_completeness.py b/tests/test_pending_order_completeness.py index e7f91c24..81ea1708 100644 --- a/tests/test_pending_order_completeness.py +++ b/tests/test_pending_order_completeness.py @@ -46,6 +46,7 @@ def accountValues(self): def execute(monkeypatch, tmp_path): # Keep the actual default-on admission, authority validator, claim, lock, # pending collector and submit wrapper. Only notification IO is replaced. + # Rule A requires an explicit fake peak equal to NAV so the original order checks exercise a measured zero drawdown. monkeypatch.delenv("ACCOUNT_NEW_RISK_GATE", raising=False) monkeypatch.setattr(service, "maybe_publish_attention_for_admission", lambda *a, **k: {}) submitted, claims = [], [] @@ -71,7 +72,7 @@ def call(mode="empty", *, profile="soxl_soxx_trend_income", reducing=False, "snapshot_as_of": None, "portfolio_equity": 1000., "targets": targets, "max_target_values": {s: 100. for s in targets}}, } - account = {"equity": 1000., "buying_power": 1000.} + account = {"peak_equity_usd": 1000., "equity": 1000., "buying_power": 1000.} if unknown_pending: account["unknown_pending_orders"] = True metadata["account_new_risk_snapshot"]["reconciliation_status"] = "UNVERIFIED" @@ -156,12 +157,14 @@ def test_complete_empty_snapshot_preserves_reductions(execute, profile): def test_claim_failure_still_prevents_delegate(execute): + # Rule A requires measured peak evidence; peak = NAV gives zero drawdown and preserves this order invariant. with pytest.raises(RuntimeError, match="execution claim required"): execute(claim=False) assert execute.submitted == [] and len(execute.claims) == 1 def test_successful_cycle_still_holds_same_day_lock(execute): + # Rule A requires measured peak evidence; peak = NAV gives zero drawdown and preserves this order invariant. execute() _, summary = execute() assert len(execute.submitted) == 1 and len(execute.claims) == 1 @@ -259,6 +262,7 @@ def test_external_pending_preserves_sell_without_reinvesting_proceeds(execute, m @pytest.mark.parametrize("profile", PROFILES) def test_other_account_external_pending_does_not_reserve_this_accounts_cash(execute, monkeypatch, profile): + # Rule A requires measured peak evidence; peak = NAV gives zero drawdown and preserves this order invariant. monkeypatch.setattr(FakeIB, "_rows", lambda self, mode: [_external_order(account="DU00000004")]) _, summary = execute("external", profile=profile) assert [(i.side, i.quantity) for i in execute.submitted] == [("buy", 1)] @@ -268,6 +272,7 @@ def test_other_account_external_pending_does_not_reserve_this_accounts_cash(exec @pytest.mark.parametrize("profile", PROFILES) @pytest.mark.parametrize("status", ["Cancelled", "ApiCancelled", "Inactive", "Filled"]) def test_terminal_external_order_does_not_withhold_cash_or_add_proceeds(execute, monkeypatch, profile, status): + # Rule A requires measured peak evidence; peak = NAV gives zero drawdown and preserves this order invariant. monkeypatch.setattr(FakeIB, "_rows", lambda self, mode: [_external_order(status=status)]) _, summary = execute("external", profile=profile) assert [(i.side, i.quantity) for i in execute.submitted] == [("buy", 1)] @@ -275,6 +280,7 @@ def test_terminal_external_order_does_not_withhold_cash_or_add_proceeds(execute, def test_external_cash_block_does_not_take_execution_lock_before_fresh_retry(execute, monkeypatch): + # Rule A requires measured peak evidence; peak = NAV gives zero drawdown and preserves this order invariant. rows = [_external_order()] monkeypatch.setattr(FakeIB, "_rows", lambda self, mode: rows) _, blocked = execute("external") diff --git a/tests/test_persistent_drawdown_breaker.py b/tests/test_persistent_drawdown_breaker.py index b2e34854..6c3e005c 100644 --- a/tests/test_persistent_drawdown_breaker.py +++ b/tests/test_persistent_drawdown_breaker.py @@ -141,6 +141,7 @@ def test_durable_open_dominates_closed_projection(): @pytest.mark.parametrize("failure", ["missing", "old_capability", "read", "write", "corruption"]) def test_persistence_failures_prohibit_new_risk(failure, tmp_path, monkeypatch): + # Rule A requires measured peak evidence; (50,000 - 40,000) / 50,000 = 0.2 exercises the actual journal failure. session = session_at(tmp_path) if failure == "missing": session = None @@ -152,9 +153,9 @@ def fail(*_args, **_kwargs): raise OSError("synthetic persistence failure") monkeypatch.setattr(AccountSubmissionSession, method, fail) else: - session.record_circuit_breaker_open(snapshot=gate.build_snapshot_from_portfolio({"total_equity": 40_000, "drawdown_from_peak": .2})) + session.record_circuit_breaker_open(snapshot=gate.build_snapshot_from_portfolio({"peak_equity_usd": 50_000, "total_equity": 40_000, "drawdown_from_peak": .2})) session._location(1).write_text("invalid json") - result = gate.evaluate_persistent_account_new_risk_admission({"equity": 40_000, "drawdown_from_peak": .2}, session=session) + result = gate.evaluate_persistent_account_new_risk_admission({"peak_equity_usd": 50_000, "equity": 40_000, "drawdown_from_peak": .2}, session=session) assert gate.new_risk_buy_prohibited(result) assert "PERSISTENT_BREAKER_UNAVAILABLE" in result.reason_codes diff --git a/tests/test_rebalance_service.py b/tests/test_rebalance_service.py index e44a9dd8..cb6485c2 100644 --- a/tests/test_rebalance_service.py +++ b/tests/test_rebalance_service.py @@ -40,8 +40,11 @@ def _run_with_synthetic_account_nlv(*, account_nlv, **kwargs): read_portfolio = kwargs.pop("get_current_portfolio") def read_snapshot(ib): - snapshot = _legacy_portfolio_snapshot(ib, get_current_portfolio=read_portfolio) + positions, values = read_portfolio(ib) + snapshot = _legacy_portfolio_snapshot(ib, get_current_portfolio=lambda _ib: (positions, values)) + # Rule A needs the explicit fake peak retained from this same snapshot read. return replace(snapshot, metadata={ + **({"peak_equity_usd": values["peak_equity_usd"]} if "peak_equity_usd" in values else {}), "broker_net_liquidation": account_nlv, "total_equity_source": "broker_net_liquidation", }) @@ -1083,6 +1086,7 @@ def disconnect(self): @pytest.fixture def claim_cycle(tmp_path, monkeypatch, strategy_module): + # Rule A requires an explicit fake peak equal to NAV so the original order checks exercise a measured zero drawdown. submitted = [] metadata = { "strategy_profile": "global_etf_rotation", @@ -1179,7 +1183,7 @@ def execute_with_observed_claim(*args, **kwargs): def run(**overrides): return _run_with_synthetic_account_nlv(account_nlv=1000.0, connect_ib=lambda: ib, - get_current_portfolio=lambda _ib: (state.positions, {"equity": 1000.0, "buying_power": 1000.0}), + get_current_portfolio=lambda _ib: (state.positions, {"peak_equity_usd": 1000.0, "equity": 1000.0, "buying_power": 1000.0}), compute_signals=lambda *_args: (state.target_weights, "synthetic", False, "", state.metadata), execute_rebalance=strategy_module.execute_rebalance, send_tg_message=lambda _message: None, @@ -1194,6 +1198,7 @@ def run(**overrides): @pytest.mark.parametrize("case", ["default", "disabled", "no_store", "empty_key", "false", "exception"]) @pytest.mark.parametrize("side", ["buy", "sell"]) def test_cycle_requires_claim_before_actual_submit(claim_cycle, case, side): + # Rule A requires measured peak evidence; peak = NAV gives zero drawdown and preserves this order invariant. state = claim_cycle if side == "sell": state.positions = {"VOO": {"quantity": 5, "avg_cost": 100.0}} @@ -1233,6 +1238,7 @@ def test_cycle_requires_claim_before_actual_submit(claim_cycle, case, side): def test_cycle_success_claim_precedes_one_submit_and_blocks_repeat(claim_cycle): + # Rule A requires measured peak evidence; peak = NAV gives zero drawdown and preserves this order invariant. state = claim_cycle from application.rebalance_service import _build_execution_marker_key @@ -1253,6 +1259,7 @@ def checked_submit(ib, intent): def test_cycle_unknown_submission_retains_claim_after_store_reopen(claim_cycle): + # Rule A requires measured peak evidence; peak = NAV gives zero drawdown and preserves this order invariant. state = claim_cycle def accepted_then_timeout(_ib, intent): @@ -1269,6 +1276,7 @@ def accepted_then_timeout(_ib, intent): def test_cycle_distinct_orders_share_one_successful_claim(claim_cycle): + # Rule A requires measured peak evidence; peak = NAV gives zero drawdown and preserves this order invariant. state = claim_cycle state.metadata["allocation"] = _weight_allocation( {"VOO": 0.3, "SPY": 0.3}, risk_symbols=("VOO", "SPY"), @@ -1282,6 +1290,7 @@ def test_cycle_distinct_orders_share_one_successful_claim(claim_cycle): @pytest.mark.parametrize("case", ["dry_run", "empty_plan", "no_signal", "admission_rejected"]) def test_cycle_non_submission_paths_do_not_claim(claim_cycle, case): + # Rule A requires measured peak evidence; peak = NAV gives zero drawdown and preserves this order invariant. state = claim_cycle store = Mock(spec=("has_marker", "claim_marker", "record_marker", "read_marker")) store.has_marker.return_value = False @@ -1387,6 +1396,7 @@ def fake_venue(_ib, intent, **_kwargs): def test_persistent_runner_first_ack_and_new_healthy_intent(persistent_runner): + # Rule A requires measured peak evidence; peak = NAV gives zero drawdown and preserves this order invariant. state = persistent_runner first = state.run() assert first.execution_summary["orders_pending"] @@ -1403,6 +1413,7 @@ def test_persistent_runner_first_ack_and_new_healthy_intent(persistent_runner): def test_persistent_runner_unknown_new_date_target_snapshot_and_healthy_injection_still_blocks(persistent_runner): + # Rule A requires measured peak evidence; peak = NAV gives zero drawdown and preserves this order invariant. state = persistent_runner state.status = "Unknown" first = state.run() diff --git a/tests/test_runtime_broker_adapters.py b/tests/test_runtime_broker_adapters.py index 28da41ce..23e85be1 100644 --- a/tests/test_runtime_broker_adapters.py +++ b/tests/test_runtime_broker_adapters.py @@ -762,6 +762,7 @@ def test_persistent_main_submit_without_grant_never_reaches_adapter(strategy_mod @pytest.fixture def persistent_real_caller(tmp_path, monkeypatch, strategy_module_factory): + # Rule A requires an explicit fake peak equal to NAV so the original order checks exercise a measured zero drawdown. """Actual main -> adapter -> executor -> platform/QPK -> fake placeOrder.""" from dataclasses import replace from functools import partial @@ -847,7 +848,7 @@ def run(*, execution_session="new", dry_run=False): state.adapters = replace(state.adapters, dry_run_only=dry_run) return module.execute_rebalance( ib, targets, {"VOO": {"quantity": 10.0, "avg_cost": 50.0}}, - {"equity": 2000.0, "buying_power": 1500.0}, signal_metadata=metadata, + {"peak_equity_usd": 2000.0, "equity": 2000.0, "buying_power": 1500.0}, signal_metadata=metadata, acquire_execution_claim=lambda: state.claim_calls.append(True) or True, account_execution_session=selected, ) @@ -858,6 +859,7 @@ def run(*, execution_session="new", dry_run=False): def test_persistent_main_real_caller_transmits_and_records_two_known_acks(persistent_real_caller): + # Rule A requires measured peak evidence; peak = NAV gives zero drawdown and preserves this order invariant. state = persistent_real_caller _logs, summary = state.run() assert [(c.symbol, o.action, o.totalQuantity) for c, o in state.submitted] == [("VOO", "SELL", 10), ("QQQM", "BUY", 20)] @@ -920,6 +922,7 @@ def test_persistent_main_old_contract_imports_but_no_submit(persistent_real_call @pytest.mark.parametrize("status,filled,price", [("PartiallyFilled", 1, 49), ("Rejected", 0, 0)]) def test_persistent_main_first_partial_or_reject_retains_healthy_path(persistent_real_caller, status, filled, price): + # Rule A requires measured peak evidence; peak = NAV gives zero drawdown and preserves this order invariant. state = persistent_real_caller state.status, state.filled, state.fill_price = status, filled, price _logs, summary = state.run() @@ -946,6 +949,7 @@ def accepted_then_disconnect(instrument, order): def test_persistent_main_old_worker_paused_at_venue_cannot_be_stolen(persistent_real_caller, monkeypatch): + # Rule A requires measured peak evidence; peak = NAV gives zero drawdown and preserves this order invariant. import threading from concurrent.futures import ThreadPoolExecutor from quant_platform_kit.common.models import OrderIntent From 76cc9172b33815e3ae595b8d50e1c50989bc5004 Mon Sep 17 00:00:00 2001 From: Codex Date: Wed, 4 Nov 2026 14:06:28 +0800 Subject: [PATCH 2/3] fix: Preserve native drawdown while refreshing cached values --- application/account_new_risk_gate_support.py | 45 ++++++++++++++------ tests/test_drawdown_native_compatibility.py | 30 +++++++++++++ 2 files changed, 61 insertions(+), 14 deletions(-) create mode 100644 tests/test_drawdown_native_compatibility.py diff --git a/application/account_new_risk_gate_support.py b/application/account_new_risk_gate_support.py index e5125bb3..a8c802f9 100644 --- a/application/account_new_risk_gate_support.py +++ b/application/account_new_risk_gate_support.py @@ -166,8 +166,6 @@ 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", "realized_vol"): if key in portfolio: projection[key] = _risk_input(portfolio[key]) if portfolio[key] is not None else "INVALID_RISK_INPUT" @@ -178,12 +176,23 @@ def build_account_new_risk_snapshot( projection.pop(_DRAWDOWN_TIMESTAMP_KEY, None) if _DRAWDOWN_TIMESTAMP_KEY in portfolio: projection[_DRAWDOWN_TIMESTAMP_KEY] = portfolio[_DRAWDOWN_TIMESTAMP_KEY] - 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), - ) + 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. @@ -210,11 +219,18 @@ def build_portfolio_from_account_values( for key in ("peak_equity_usd", "drawdown_from_peak", "realized_vol", _DRAWDOWN_TIMESTAMP_KEY): if key in account_values: portfolio[key] = account_values[key] - elif key in metadata: - portfolio[key] = metadata[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: @@ -228,10 +244,11 @@ def build_portfolio_from_account_values( "account_new_risk_snapshot": projection, } ) - # Both native and top-level copy paths expose the same revalidated value. - portfolio["drawdown_from_peak"] = portfolio["account_new_risk_snapshot"]["drawdown_from_peak"] - if _DRAWDOWN_TIMESTAMP_KEY in portfolio["account_new_risk_snapshot"]: - portfolio[_DRAWDOWN_TIMESTAMP_KEY] = portfolio["account_new_risk_snapshot"][_DRAWDOWN_TIMESTAMP_KEY] + 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 diff --git a/tests/test_drawdown_native_compatibility.py b/tests/test_drawdown_native_compatibility.py new file mode 100644 index 00000000..ffd27f82 --- /dev/null +++ b/tests/test_drawdown_native_compatibility.py @@ -0,0 +1,30 @@ +"""Keep native account drawdown distinct from cached strategy projections.""" +import math +import struct + +import pytest + +from application import account_new_risk_gate_support as gate +from tests.test_drawdown_cache_postmerge import facts, projected + + +@pytest.mark.parametrize("drawdown", [ + 0.0, .074999, math.nextafter(.075, 0), .075, + math.nextafter(.075, 1), .075001, .10, .15, .20, +]) +@pytest.mark.parametrize("cached", [0.0, .6]) +def test_native_explicit_drawdown_keeps_main_bits(drawdown, cached): + values, metadata = facts("native-snapshot", cached=cached) + values.update(drawdown_from_peak=drawdown, + peak_equity_usd=40_000 / (1 - drawdown)) + snapshot = projected(values, metadata) + # Main passes through the explicit native float, without an inverse round-trip. + assert struct.pack("!d", snapshot.drawdown_from_peak) == struct.pack("!d", drawdown) + + +@pytest.mark.parametrize("variant", ["native-snapshot", "top-level-copy"]) +def test_rebuilt_cached_projection_uses_current_nav(variant): + values, metadata = facts(variant, cached=0.0) + portfolio = gate.build_portfolio_from_account_values(values, signal_metadata=metadata) + portfolio["total_equity"] = 50_000 + assert gate.build_snapshot_from_portfolio(portfolio).drawdown_from_peak == 0.0 From 9e68fdaf97048dc66e8da185fe2f1cb8733446a3 Mon Sep 17 00:00:00 2001 From: Codex Date: Wed, 4 Nov 2026 16:56:46 +0800 Subject: [PATCH 3/3] fix: Preserve drawdown observation time in portfolio views --- application/rebalance_service.py | 2 +- tests/test_rebalance_service.py | 29 +++++++++++++++++++++++++++++ 2 files changed, 30 insertions(+), 1 deletion(-) diff --git a/application/rebalance_service.py b/application/rebalance_service.py index 9b337364..dac63884 100644 --- a/application/rebalance_service.py +++ b/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] diff --git a/tests/test_rebalance_service.py b/tests/test_rebalance_service.py index cb6485c2..bd2d098d 100644 --- a/tests/test_rebalance_service.py +++ b/tests/test_rebalance_service.py @@ -77,6 +77,35 @@ def test_heartbeat_uses_broker_total_without_changing_strategy_equity(): assert account_values["heartbeat_account_snapshot"]["available_cash"] == 100.0 +@pytest.mark.parametrize("observed_at", [ + "2026-11-04T00:00:00+00:00", "2020-01-01T00:00:00Z", None, "invalid", +]) +def test_snapshot_portfolio_preserves_drawdown_observation_evidence(observed_at): + metadata = { + "broker_net_liquidation": 40000.0, + "total_equity_source": "broker_net_liquidation", + "drawdown_from_peak": 0.2, + "drawdown_from_peak_observed_at": observed_at, + } + snapshot = SimpleNamespace(positions=(), total_equity=250.0, metadata=metadata.copy()) + + _positions, account_values = _snapshot_to_portfolio_view(snapshot) + + assert "drawdown_from_peak_observed_at" in account_values + assert account_values["drawdown_from_peak_observed_at"] == observed_at + assert account_values["drawdown_from_peak"] == 0.2 + assert account_values["account_equity_usd"] == 40000.0 + assert snapshot.metadata == metadata + + +def test_snapshot_portfolio_does_not_invent_drawdown_observation_time(): + snapshot = SimpleNamespace(positions=(), metadata={"drawdown_from_peak": 0.2}) + + _positions, account_values = _snapshot_to_portfolio_view(snapshot) + + assert "drawdown_from_peak_observed_at" not in account_values + + def test_no_trade_heartbeat_respects_window_and_notification_policy(): idle = {"actionable": False, "risk_flags": ()} assert not _should_suppress_noop_notification(idle, order_count=0, notify_no_trade_cycles=True)