diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 67d0a9e9..fd864b83 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -101,7 +101,7 @@ jobs: - name: Check QPK pin consistency run: | set -euo pipefail - QPK_EXPECTED_PIN=6136b1c667a60d484f500c6d6046c0585c75305d uv run --no-sync python scripts/check_qpk_pin_consistency.py + QPK_EXPECTED_PIN=cee53965b3d6b368445193b580aa94f2bbb9537c uv run --no-sync python scripts/check_qpk_pin_consistency.py - name: Ensure uv.lock matches pyproject.toml run: uv lock --check diff --git a/application/account_new_risk_gate_support.py b/application/account_new_risk_gate_support.py index 6f279d78..b01fe6cd 100644 --- a/application/account_new_risk_gate_support.py +++ b/application/account_new_risk_gate_support.py @@ -7,6 +7,7 @@ import os from collections.abc import Mapping from contextlib import contextmanager +from dataclasses import replace from typing import Any from quant_platform_kit.risk.account_new_risk_gate import ( @@ -42,6 +43,14 @@ def is_account_new_risk_gate_enabled() -> bool: return str(os.environ.get(ACCOUNT_NEW_RISK_GATE_ENV, "") or "").strip() != "0" +def is_portfolio_loss_stoploss_enabled() -> bool: + """Read the existing switch; absent or malformed values confer no exit authority.""" + try: + return str(os.environ.get("PORTFOLIO_LOSS_STOPLOSS_ENABLED", "")).strip().lower() in {"1", "true", "yes", "on"} + except Exception: + return False + + def _coerce_optional_float(value: object) -> float | None: if value is None or isinstance(value, bool): return None @@ -75,10 +84,10 @@ def _mapping_or_empty(value: object) -> Mapping[str, Any]: def _resolve_equity_usd(portfolio: Mapping[str, Any], execution: Mapping[str, Any] | None) -> float | None: - for key in ("total_equity", "total_strategy_equity", "equity"): - equity = _coerce_optional_float(portfolio.get(key)) - if equity is not None and equity > 0.0: - return equity + for key in ("account_equity_usd", "total_equity", "total_strategy_equity", "equity"): + if key in portfolio: + equity = _coerce_optional_float(portfolio[key]) + return equity if equity is not None and equity > 0.0 else None if execution is not None: equity = _coerce_optional_float(execution.get("portfolio_total_equity")) if equity is not None and equity > 0.0: @@ -92,20 +101,22 @@ def _portfolio_metadata(portfolio: Mapping[str, Any]) -> Mapping[str, Any]: def _resolve_unknown_pending(portfolio: Mapping[str, Any], projection: Mapping[str, Any]) -> bool: - """Only an explicit unknown-pending signal counts; absence is never treated as pending.""" - if "unknown_pending_orders" in projection: - return bool(projection.get("unknown_pending_orders")) - if portfolio.get("unknown_pending_orders") is not None: - return bool(portfolio.get("unknown_pending_orders")) - return bool(_portfolio_metadata(portfolio).get("unknown_pending_orders")) + """Any explicit unsafe or unknown signal blocks, regardless of its source.""" + return any( + source["unknown_pending_orders"] is not False + for source in (portfolio, _portfolio_metadata(portfolio), projection) + if "unknown_pending_orders" in source + ) def _resolve_durable_breaker_open(portfolio: Mapping[str, Any], projection: Mapping[str, Any]) -> bool: """Only an explicit durable OPEN state trips the breaker; absence is never OPEN.""" for source in (projection, portfolio, _portfolio_metadata(portfolio)): - value = source.get("durable_circuit_breaker_state") - if value is not None: - return str(value).strip().upper() == "OPEN" + if "durable_circuit_breaker_state" not in source: + continue + value = source["durable_circuit_breaker_state"] + if not isinstance(value, str) or value.strip().upper() != "CLOSED": + return True return False @@ -116,30 +127,46 @@ def build_account_new_risk_snapshot( ) -> dict[str, Any]: """Build a cycle-health projection from evidence available this cycle. - ``observation_ok`` reflects whether equity resolved this cycle; - ``unknown_pending`` and ``durable_breaker_open`` only trip on an explicit - signal from the portfolio or its ``metadata`` -- their absence is never - mistaken for a durable OPEN breaker or pending reconciliation. Explicit - keys already present on an injected ``account_new_risk_snapshot`` always - win over the projected axes (tests / HITL overrides). + Explicit unsafe native facts and failed cycle evidence cannot be overridden + by a strategy projection. Missing optional axes remain optional; supplied + invalid values remain invalid for the downstream fail-closed validator. """ projection = dict(portfolio.get("account_new_risk_snapshot") or {}) - equity_usd = _coerce_optional_float(projection.get("equity_usd")) - if equity_usd is None: + native_equity = next((key for key in ( + "account_equity_usd", "total_equity", "total_strategy_equity", "equity", + ) if key in portfolio), None) + equity_usd = (_coerce_optional_float(portfolio[native_equity]) if native_equity + else _coerce_optional_float(projection.get("equity_usd"))) + if native_equity is None and "equity_usd" not in projection: equity_usd = _resolve_equity_usd(portfolio, execution) + if equity_usd is not None and equity_usd <= 0: + equity_usd = None evidence = CycleNewRiskHealthEvidence( observation_ok=equity_usd is not None, unknown_pending=_resolve_unknown_pending(portfolio, projection), durable_breaker_open=_resolve_durable_breaker_open(portfolio, projection), ) projection = apply_cycle_new_risk_health_axes(projection, evidence) + if not evidence.observation_ok: + projection["observation_status"] = "UNAVAILABLE" + if evidence.unknown_pending: + projection["reconciliation_status"] = "UNVERIFIED" + if evidence.durable_breaker_open or evidence.unknown_pending: + projection["circuit_breaker_state"] = "OPEN" + for key, healthy in (("observation_status", "COMPLETE"), + ("reconciliation_status", "VERIFIED"), + ("circuit_breaker_state", "CLOSED")): + 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"): - if key in projection: + 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]) - elif key in portfolio: - projection[key] = _risk_input(portfolio[key]) return projection @@ -151,21 +178,33 @@ def build_portfolio_from_account_values( """Project a portfolio dict from an existing IBKR account_values read.""" metadata = dict(signal_metadata or {}) portfolio: dict[str, Any] = { - "total_equity": _coerce_optional_float(account_values.get("equity")), + "total_equity": _coerce_optional_float(account_values.get("account_equity_usd", account_values.get("equity"))), } - for key in ("unknown_pending_orders", "durable_circuit_breaker_state"): + for key in ("unknown_pending_orders", "durable_circuit_breaker_state", + "observation_status", "reconciliation_status", "circuit_breaker_state", "production_drift_status", "daily_loss_usd"): if key in 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"): + if key in account_values: + portfolio[key] = account_values[key] + 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"] + if "peak_equity_usd" in account_values and "drawdown_from_peak" not in account_values: + projection.pop("drawdown_from_peak", None) + for key in ("peak_equity_usd", "drawdown_from_peak"): + if key in account_values: + projection[key] = account_values[key] portfolio["account_new_risk_snapshot"] = build_account_new_risk_snapshot( { **portfolio, - "account_new_risk_snapshot": metadata.get("account_new_risk_snapshot"), + "account_new_risk_snapshot": projection, } ) for key in ("peak_equity_usd", "drawdown_from_peak", "realized_vol"): - if key in metadata: + if key not in portfolio and key in metadata: portfolio[key] = metadata[key] return portfolio @@ -192,7 +231,7 @@ def _resolve_production_drift_status( portfolio: Mapping[str, Any], projection: Mapping[str, Any] ) -> str | None: """Prefer explicit inject; else read-only PerformanceStore (Policy A). Fail-soft on store errors.""" - for source in (projection, portfolio, _portfolio_metadata(portfolio)): + for source in (portfolio, projection, _portfolio_metadata(portfolio)): raw = source.get("production_drift_status") if raw is not None and raw != "": return str(raw).strip() @@ -221,7 +260,7 @@ def _resolve_explicit_daily_loss_usd( execution: Mapping[str, Any] | None, ) -> float | None: """Pass through an explicit daily_loss_usd fact only; never invent one.""" - for source in (projection, portfolio, _mapping_or_empty(execution)): + for source in (portfolio, projection, _mapping_or_empty(execution)): if "daily_loss_usd" in source: return _coerce_optional_float(source.get("daily_loss_usd")) return None @@ -300,8 +339,6 @@ def build_snapshot_from_portfolio( """Project an injected reconciliation snapshot from an existing portfolio read.""" projection = build_account_new_risk_snapshot(portfolio, execution=execution) equity_usd = _coerce_optional_float(projection.get("equity_usd")) - if equity_usd is None: - equity_usd = _resolve_equity_usd(portfolio, execution) peak_equity_usd = _risk_input(projection.get("peak_equity_usd")) explicit_dd = _risk_input(projection.get("drawdown_from_peak")) return InjectedReconciliationSnapshot( @@ -346,6 +383,55 @@ def new_risk_buy_prohibited(result: NewRiskAdmissionResult) -> bool: return result.disposition == NewRiskDisposition.NEW_RISK_PROHIBITED +def evaluate_persistent_account_new_risk_admission( + account_values: Mapping[str, Any], *, signal_metadata=None, session=None, +) -> NewRiskAdmissionResult: + """Restore OPEN before admission and record a breach before any cycle return. + + The existing physical-account journal is the authority. Missing capability, + corruption, conflicting creates and failed readbacks prohibit new risk. + There is deliberately no reset consumer or recovery-based clear. + """ + portfolio = build_portfolio_from_account_values(account_values, signal_metadata=signal_metadata) + admission = evaluate_portfolio_new_risk_admission(portfolio) + try: + from quant_platform_kit.common.execution_state import AccountSubmissionSession + if not isinstance(session, AccountSubmissionSession) or not callable(getattr(session, "record_circuit_breaker_open", None)): + raise RuntimeError("persistent breaker capability unavailable") + if admission.hard_brake_triggered: + session.record_circuit_breaker_open(snapshot=build_snapshot_from_portfolio(portfolio)) + state = session.read_state() + if type(state.get("circuit_breaker_open")) is not bool: + raise RuntimeError("persistent breaker readback unavailable") + if not state["circuit_breaker_open"]: + return admission + reason = "CIRCUIT_BREAKER_OPEN" + except Exception: + reason = "PERSISTENT_BREAKER_UNAVAILABLE" + return replace( + admission, disposition=NewRiskDisposition.NEW_RISK_PROHIBITED, + reason_codes=tuple(dict.fromkeys((*admission.reason_codes, reason))), + ) + + +def scale_targets_for_admission(target_weights, admission, *, snapshot=None): + """Exclude drawdown sizing without exit authority; retain capital/vol sizing.""" + scale = admission.combined_scale + if not is_portfolio_loss_stoploss_enabled(): + # Keep the original admission and persisted OPEN authoritative for buys. + # Reuse the same cycle facts, omitting only the drawdown axis for targets. + scale = None + if snapshot is not None: + try: + sizing = evaluate_new_risk_admission(replace( + snapshot, peak_equity_usd=None, drawdown_from_peak=None, + )) + scale = sizing.combined_scale + except AccountNewRiskGateError: + pass # Unknown sizing cannot confer exit authority. + return apply_combined_scale_to_target_weights(target_weights, scale) + + def apply_combined_scale_to_target_weights( target_weights: Mapping[str, Any] | None, combined_scale: float | None, diff --git a/application/execution_service.py b/application/execution_service.py index 7125baa0..ffb277f0 100644 --- a/application/execution_service.py +++ b/application/execution_service.py @@ -16,10 +16,11 @@ import pandas as pd from application.account_new_risk_gate_support import ( - apply_combined_scale_to_target_weights, + scale_targets_for_admission, build_portfolio_from_account_values, build_snapshot_from_portfolio, evaluate_account_values_new_risk_admission, + evaluate_persistent_account_new_risk_admission, is_account_new_risk_gate_enabled, maybe_publish_attention_for_admission, new_risk_buy_prohibited, @@ -1452,10 +1453,16 @@ def submit_claimed_order(ib, order_intent, **kwargs): raise RuntimeError("IBKR execution claim required; refusing broker submission") # Shadow consult after successful claim; deny must not block submit. consult_t1_live_submit(identity_held=True, dry_run_bypass=False) - # The venue wrapper finishes blocking preparation and resizing before - # reserving the exact cash-buy intent. Other paths keep their grant. + new_risk = str(order_intent.side).strip().lower() == "buy" and is_account_new_risk_gate_enabled() + if new_risk: + latest = evaluate_persistent_account_new_risk_admission( + account_values, signal_metadata=signal_metadata, session=account_execution_session, + ) + if new_risk_buy_prohibited(latest): + raise RuntimeError("IBKR persistent account breaker prohibits new risk") + # Settlement preparation must finish before reserving the exact intent. grant = (None if isinstance(ib, _SettledCashDeadlineIB) - else account_execution_session.reserve(order_intent)) + else account_execution_session.reserve(order_intent, new_risk=new_risk)) return delegate_submit( ib, order_intent, account_execution_session=account_execution_session, execution_grant=grant, **kwargs, @@ -1572,10 +1579,16 @@ def record_quote_snapshot(symbol, snapshot) -> None: account_new_risk_buy_blocked = False account_new_risk_reason_codes: tuple[str, ...] = () if is_account_new_risk_gate_enabled(): - admission = evaluate_account_values_new_risk_admission( - account_values, - signal_metadata=signal_metadata, - ) + # Inner dry-run previews without a session retain their diagnostic + # purpose; they make no persistence claim. Native cycles always restore + # the journal, including no-target and dry-run cycles with a store. + admission = evaluate_account_values_new_risk_admission(account_values, signal_metadata=signal_metadata) + if dry_run_only and account_execution_session is None: + execution_summary["persistent_breaker_status"] = "NOT_PROVEN" + else: + admission = evaluate_persistent_account_new_risk_admission( + account_values, signal_metadata=signal_metadata, session=account_execution_session, + ) account_new_risk_buy_blocked = new_risk_buy_prohibited(admission) account_new_risk_reason_codes = tuple(admission.reason_codes) portfolio = build_portfolio_from_account_values(account_values, signal_metadata=signal_metadata) @@ -1593,14 +1606,11 @@ def record_quote_snapshot(symbol, snapshot) -> None: f"failed={attention_counts.get('failed', 0)}", flush=True, ) - target_weights = apply_combined_scale_to_target_weights( - target_weights, - admission.combined_scale, - ) + target_weights = scale_targets_for_admission(target_weights, admission, snapshot=cycle_snapshot) if admission.combined_scale is not None: print( f"[Envelope scale] combined_scale={admission.combined_scale} " - "applied_to_target_weights", + "subject_to_stoploss_exit_authority", flush=True, ) else: @@ -2020,6 +2030,10 @@ def cash_sweep_sale_quantity_to_fund_buy(max_quantity: int, candidate_symbols: t break quantity_zero_symbols.append(symbol) + if (has_buy_plan and not dry_run_only + and "PERSISTENT_BREAKER_UNAVAILABLE" in account_new_risk_reason_codes): + raise RuntimeError("IBKR persistent breaker session/contract unavailable; refusing new risk") + if ( (pending_observation.cash_risk_unknown or settled_cash_ledger_unverified) and any(target_mv[s] > current_mv.get(s, 0.0) + threshold for s in target_mv) diff --git a/application/rebalance_service.py b/application/rebalance_service.py index 404a6236..8d92d344 100644 --- a/application/rebalance_service.py +++ b/application/rebalance_service.py @@ -706,9 +706,19 @@ 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"): + 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", + "production_drift_status", "daily_loss_usd"): if key in metadata: account_values[key] = metadata[key] + # Strategy equity is for sizing, not physical-account risk admission. + # Keep an explicit unavailable value so neither sizing equity nor an older + # healthy strategy projection can substitute for unqualified account NAV. + account_values["account_equity_usd"] = ( + metadata.get("broker_net_liquidation") + if metadata.get("total_equity_source") == "broker_net_liquidation" + else None + ) return positions, account_values @@ -733,6 +743,7 @@ def _strategy_portfolio_view(positions, account_values, strategy_symbols): buying_power = float(dict(account_values or {}).get("buying_power") or 0.0) filtered_account_values = { **dict(account_values or {}), + "account_equity_usd": dict(account_values or {}).get("account_equity_usd", dict(account_values or {}).get("equity")), "equity": buying_power + strategy_market_value, } return filtered_positions, filtered_account_values @@ -990,16 +1001,55 @@ def run_strategy_core( else: target_weights, signal_desc, _is_emergency, status_desc = signal_result signal_metadata = {} + account_execution_session = None + if getattr(config, "execution_state_store", None) is not None: + from quant_platform_kit.common import execution_state as contract + + session_type = getattr(contract, "AccountSubmissionSession", None) + if (session_type is not None and getattr(config, "execution_state_store", None) is not None + and getattr(session_type, "contract_version", None) == "account_submission.v1"): + try: + profile = str(signal_metadata.get("strategy_profile") or config.strategy_profile or "") + # A no-target cycle has no approved order intent. Its + # observation-only identity cannot pass the executor's + # allocation/intent binding, but can record account OPEN. + intent_id = ( + contract.submission_intent_digest(signal_metadata, profile) + if target_weights is not None else contract.submission_binding_digest( + {"purpose": "account_breaker_observation", "profile": profile} + ) + ) + account_execution_session = session_type( + config.execution_state_store, broker="ibkr", + account_id=_resolve_physical_account_id(config=config, snapshot=snapshot), + owner_id=profile, profile=profile, + intent_id=intent_id, + release_id=contract.submission_release_digest(config.runtime_release_receipt), + ) + except (RuntimeError, ValueError, TypeError): + # Never fabricate a healthy account root or legacy grant. + account_execution_session = None + + from application.account_new_risk_gate_support import ( + evaluate_persistent_account_new_risk_admission, + is_account_new_risk_gate_enabled, + is_portfolio_loss_stoploss_enabled, + ) + signal_metadata = dict(signal_metadata or {}) + admission = evaluate_persistent_account_new_risk_admission( + account_values, signal_metadata=signal_metadata, session=account_execution_session, + ) + if is_account_new_risk_gate_enabled(): + if "CIRCUIT_BREAKER_OPEN" in admission.reason_codes: + account_values["durable_circuit_breaker_state"] = "OPEN" if target_weights is None: - from application.account_new_risk_gate_support import evaluate_account_values_new_risk_admission - admission = evaluate_account_values_new_risk_admission(account_values, signal_metadata=signal_metadata) signal_metadata = dict(signal_metadata or {}) signal_metadata["account_brake_observation"] = { "hard_brake_triggered": bool(getattr(admission, "hard_brake_triggered", False)), "reason_codes": list(admission.reason_codes), } if signal_metadata["account_brake_observation"]["hard_brake_triggered"]: - if (getattr(config, "dry_run_only", False) or signal_metadata.get("dry_run_only") is True or account_brake_session is None + if (not is_portfolio_loss_stoploss_enabled() or getattr(config, "dry_run_only", False) or signal_metadata.get("dry_run_only") is True or account_brake_session is None or account_brake_submit is None or account_brake_order_intent_cls is None): signal_metadata["execution_blocked_reason"] = "account_brake_source_or_submit_unavailable" else: @@ -1368,30 +1418,6 @@ def acquire_execution_claim(): raise RuntimeError("IBKR execution claim unavailable; refusing broker submission") from None return execution_claim_acquired - # New QPK APIs are optional at import time because the declared pin is - # unchanged. Missing capability/identity leaves session=None: only a - # real submit is rejected by the executor, while no-order/dry-run stays - # free of account reservations. - account_execution_session = None - if not config.dry_run_only: - from quant_platform_kit.common import execution_state as contract - - session_type = getattr(contract, "AccountSubmissionSession", None) - if (session_type is not None and execution_state_store is not None - and getattr(session_type, "contract_version", None) == "account_submission.v1"): - try: - profile = str(signal_metadata.get("strategy_profile") or config.strategy_profile or "") - account_execution_session = session_type( - execution_state_store, broker="ibkr", - account_id=_resolve_physical_account_id(config=config, snapshot=snapshot), - owner_id=profile, profile=profile, - intent_id=contract.submission_intent_digest(signal_metadata, profile), - release_id=contract.submission_release_digest(config.runtime_release_receipt), - ) - except (RuntimeError, ValueError, TypeError): - # Never fabricate a healthy account root or legacy grant. - account_execution_session = None - execution_result = runtime.execute_rebalance( ib, resolved_target_weights, diff --git a/docs/account-capital-policy.zh-CN.md b/docs/account-capital-policy.zh-CN.md index 21147351..a2ae7e30 100644 --- a/docs/account-capital-policy.zh-CN.md +++ b/docs/account-capital-policy.zh-CN.md @@ -145,3 +145,36 @@ 设计参考：[QuantConnect 的仓位计算与现金缓冲](https://www.quantconnect.com/docs/v2/writing-algorithms/trading-and-orders/position-sizing)、 [Investor.gov 的投资期限与风险承受能力说明](https://www.investor.gov/introduction-investing/getting-started/asset-allocation)。 参考资料提供设计背景，不把其他系统的默认数字作为本项目用户政策。 + +## 7. 回撤熔断的新增风险边界 + +原生快照中的 `drawdown_from_peak` 和 `peak_equity_usd` 传到新增风险门；策略子组合 +使用独立的 `equity` 计算配置，账户回撤继续使用 `account_equity_usd`，不把子组合权益 +误当作账户净值。快照事实优先于策略元数据中的旧投影，非法值保留为错误输入。 + +原生快照转换只在 `total_equity_source` 为 `broker_net_liquidation` 时， +将已核 `broker_net_liquidation` 传为 `account_equity_usd`；缺失或未核的值明确 +标记不可用，无效值由风险门拒绝，不用 `total_equity` 或策略缓存补齐。 +转换同时保留显式的 `realized_vol`、`observation_status`、`reconciliation_status`、 +`circuit_breaker_state` 和 `production_drift_status`，确保这些当前事实到达风险门。 + +账户回撤触线后，复用 `AccountSubmissionSession` 的物理账户追加日志写入 +`CIRCUIT_BREAKER_OPEN`，包括没有策略目标而提前返回的周期。后续周期、其他策略及 +新进程读取同一账户日志；净值恢复、显式 CLOSED 或调用方的 reset 标记都不能清除 OPEN。 +本次不提供复位入口。缺少日志能力、读写失败、冲突或读回失败均禁止新增风险。 + +`PORTFOLIO_LOSS_STOPLOSS_ENABLED` 为 OFF 时，回撤熔断只阻止新建仓和加仓，目标缩放 +排除回撤分量，不因半阈值或全阈值回撤缩减持仓或产生额外卖单；无目标周期也不启动回撤减仓。 +capital 和 volatility 分量继续按原有规则缩放目标。 +ON 时继续使用原有缩放及减仓路径，原账户、订单、授权与 UNKNOWN 保护继续生效。 +开关默认值、风险参数和配置保持原样。当前分支的 QPK 固定版本为 +`cee53965b3d6b368445193b580aa94f2bbb9537c`，由 `pyproject.toml`、`qsl.toml`、 +`uv.lock` 及 CI 预期版本共同确认。 + +IBKR 的真实买单将 `new_risk=True` 绑定到日志中的 PREPARED，最终 dispatch 再检查 OPEN。 +普通已表达的减仓仍走原有证明及订单保护，持久化 OPEN 本身不授予卖出权限。 +没有账户会话的内部 dry-run 仅提供计划预览，标记持久性 `NOT_PROVEN`；原生周期会在 +执行前尝试恢复账户日志。旧依赖没有所需能力时，真实新增风险被阻止，不能将预览当成 +持久化或运行验收。当前分支已采用上述 QPK 版本；生产采用和部署仍需另行审核。 + +当前账户读取中的未知挂单、过期或未知观察状态、未核对状态和无效风险数值必须阻止新增风险；策略缓存中的健康值不能覆盖这些事实。当前账户净值缺失时，不使用策略缓存净值恢复准入。此规则不授予止损退出权限。 diff --git a/main.py b/main.py index 1473c5d9..7d1bde20 100644 --- a/main.py +++ b/main.py @@ -617,7 +617,7 @@ def submit_market_order_intent( raw_payload={"detail": "settled_cash_changed_before_reservation"}, ) order_intent = prepared - execution_grant = account_execution_session.reserve(order_intent) + execution_grant = account_execution_session.reserve(order_intent, new_risk=True) return account_execution_session.dispatch( execution_grant, order_intent, lambda: submit_order_intent( diff --git a/pyproject.toml b/pyproject.toml index b3eaacd0..df1cf258 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -21,7 +21,7 @@ dependencies = [ "google-cloud-secret-manager", "google-cloud-storage", "yfinance", - "quant-platform-kit @ git+https://github.com/QuantStrategyLab/QuantPlatformKit.git@6136b1c667a60d484f500c6d6046c0585c75305d", + "quant-platform-kit @ git+https://github.com/QuantStrategyLab/QuantPlatformKit.git@cee53965b3d6b368445193b580aa94f2bbb9537c", "us-equity-strategies @ git+https://github.com/QuantStrategyLab/UsEquityStrategies.git@3f112f5352efe677a0c480c4ab621b8ea3e85336", "hk-equity-strategies @ git+https://github.com/QuantStrategyLab/HkEquityStrategies.git@fe0c32e09d454ca4388c3372bf3874097b1bbaf0", ] @@ -64,6 +64,6 @@ include = [ [tool.uv] override-dependencies = [ - "quant-platform-kit @ git+https://github.com/QuantStrategyLab/QuantPlatformKit.git@6136b1c667a60d484f500c6d6046c0585c75305d", + "quant-platform-kit @ git+https://github.com/QuantStrategyLab/QuantPlatformKit.git@cee53965b3d6b368445193b580aa94f2bbb9537c", ] diff --git a/qsl.toml b/qsl.toml index fad5a643..4408419a 100644 --- a/qsl.toml +++ b/qsl.toml @@ -5,7 +5,7 @@ upgrade_ring = "ring_d" allow_legacy = false [qsl.requires] -quant_platform_kit = "6136b1c667a60d484f500c6d6046c0585c75305d" +quant_platform_kit = "cee53965b3d6b368445193b580aa94f2bbb9537c" us_equity_strategies = "3f112f5352efe677a0c480c4ab621b8ea3e85336" hk_equity_strategies = "709e5e1cde7841aed538d94eb26b552b46cb7806" diff --git a/tests/fixtures/notif_main_2f16cb9.json b/tests/fixtures/notif_main_2f16cb9.json index 6d1f2b46..42cb2be3 100644 --- a/tests/fixtures/notif_main_2f16cb9.json +++ b/tests/fixtures/notif_main_2f16cb9.json @@ -1810,7 +1810,7 @@ "attention:blocked:zh": { "sends": [ { - "text": "🚨 IBKR · -account · 半导体趋势收益\n结果：已禁止新增风险\n原因：operational_uncertain: 运行状态不确定 / 故障; new_risk_prohibited: 风控已拦截；调用方未提供具体原因码\n下一步：打开管理站处理恢复 / 复位（需授权）", + "text": "🚨 IBKR · ••nt · 半导体趋势收益\n结果：已禁止新增风险\n原因：operational_uncertain: 运行状态不确定 / 故障; new_risk_prohibited: 风控已拦截；调用方未提供具体原因码\n运行时间：未知\n运行详情：未提供运行链接\n下一步：打开管理站处理恢复 / 复位（需授权）", "alert_key": "attention/ibkr/-account/soxl_soxx_trend_income/halt/operational_uncertain" } ], @@ -1830,7 +1830,7 @@ "daily_digest:attention:blocked:zh": { "sends": [ { - "text": "🚨 IBKR · -account · 半导体趋势收益\n结果：已禁止新增风险\n原因：operational_uncertain: 运行状态不确定 / 故障; new_risk_prohibited: 风控已拦截；调用方未提供具体原因码\n下一步：打开管理站处理恢复 / 复位（需授权）", + "text": "🚨 IBKR · ••nt · 半导体趋势收益\n结果：已禁止新增风险\n原因：operational_uncertain: 运行状态不确定 / 故障; new_risk_prohibited: 风控已拦截；调用方未提供具体原因码\n运行时间：未知\n运行详情：未提供运行链接\n下一步：打开管理站处理恢复 / 复位（需授权）", "alert_key": "attention/ibkr/-account/soxl_soxx_trend_income/halt/operational_uncertain" } ], @@ -1850,7 +1850,7 @@ "attention:blocked:en": { "sends": [ { - "text": "🚨 IBKR · -account · Semiconductor Trend Income\nResult: new risk additions prohibited\nReason: operational_uncertain: operational uncertainty / fault; new_risk_prohibited: a risk guard blocked additions; the caller did not provide the specific reason code\nNext: open the management console for recovery / reset (authorization required)", + "text": "🚨 IBKR · ••nt · Semiconductor Trend Income\nResult: new risk additions prohibited\nReason: operational_uncertain: operational uncertainty / fault; new_risk_prohibited: a risk guard blocked additions; the caller did not provide the specific reason code\nRun time: Unknown\nRun details: Run link not provided\nNext: open the management console for recovery / reset (authorization required)", "alert_key": "attention/ibkr/-account/soxl_soxx_trend_income/halt/operational_uncertain" } ], @@ -1870,7 +1870,7 @@ "daily_digest:attention:blocked:en": { "sends": [ { - "text": "🚨 IBKR · -account · Semiconductor Trend Income\nResult: new risk additions prohibited\nReason: operational_uncertain: operational uncertainty / fault; new_risk_prohibited: a risk guard blocked additions; the caller did not provide the specific reason code\nNext: open the management console for recovery / reset (authorization required)", + "text": "🚨 IBKR · ••nt · Semiconductor Trend Income\nResult: new risk additions prohibited\nReason: operational_uncertain: operational uncertainty / fault; new_risk_prohibited: a risk guard blocked additions; the caller did not provide the specific reason code\nRun time: Unknown\nRun details: Run link not provided\nNext: open the management console for recovery / reset (authorization required)", "alert_key": "attention/ibkr/-account/soxl_soxx_trend_income/halt/operational_uncertain" } ], diff --git a/tests/test_breaker_settlement_handoff.py b/tests/test_breaker_settlement_handoff.py new file mode 100644 index 00000000..7a6f5f7a --- /dev/null +++ b/tests/test_breaker_settlement_handoff.py @@ -0,0 +1,92 @@ +"""Merged breaker/settlement boundaries with synthetic journals and broker IO.""" +import multiprocessing +import os + +import pytest + +from application.account_new_risk_gate_support import build_snapshot_from_portfolio +from tests import test_settled_cash_containment as containment +from tests.test_settled_cash_containment import FakeIB + +native = containment.native + + +@pytest.fixture(autouse=True) +def isolate_startup_environment(): + # The reused startup fixture installs inert values directly in os.environ. + keys = ("ACCOUNT_GROUP", "IB_ACCOUNT_GROUP_CONFIG_JSON", + "IB_ACCOUNT_GROUP_CONFIG_SECRET_NAME", "RUNTIME_TARGET_JSON") + previous = {key: os.environ.get(key) for key in keys} + yield + for key, value in previous.items(): + if value is None: + os.environ.pop(key, None) + else: + os.environ[key] = value + + +def _open_in_another_process(session): + session.record_circuit_breaker_open(snapshot=build_snapshot_from_portfolio({ + "total_equity": 1600., "peak_equity_usd": 2000., "drawdown_from_peak": .2, + })) + + +@pytest.mark.parametrize("phase", ["qualification", "after_reservation"]) +def test_other_process_open_blocks_deferred_cash_buy(native, monkeypatch, phase): + from quant_platform_kit.common.execution_state import AccountSubmissionSession + + plan = native({"SOXL": 400.}, cash=500.) + ib = FakeIB(cash=500., settled=500.) + + def open_breaker(): + writer = multiprocessing.get_context("fork").Process( + target=_open_in_another_process, args=(plan.session(),), + ) + writer.start() + writer.join(10) + assert writer.exitcode == 0 + + if phase == "qualification": + qualify = ib.qualifyContracts + + def qualify_then_open(contract): + prepared = qualify(contract) + open_breaker() + return prepared + + ib.qualifyContracts = qualify_then_open + else: + reserve = AccountSubmissionSession.reserve + + def reserve_then_open(session, order, **kwargs): + grant = reserve(session, order, **kwargs) + open_breaker() + return grant + + monkeypatch.setattr(AccountSubmissionSession, "reserve", reserve_then_open) + + with pytest.raises(RuntimeError, match="circuit breaker"): + plan.run(ib, prices={"SOXL": 25.}) + assert ib.orders == [] + restarted = plan.session().read_state() + assert restarted["circuit_breaker_open"] is True + assert len(ib.accountValueEvent) == 0 + + +def test_deferred_buy_marks_resized_exact_intent_as_new_risk(native): + from tests.test_settled_cash_intervening_updates import emit_settlement + + plan = native({"SOXL": 400.}, cash=500.) + ib = FakeIB(cash=500., settled=500.) + qualify = ib.qualifyContracts + + def reduce_budget(contract): + emit_settlement(ib, "100") + return qualify(contract) + + ib.qualifyContracts = reduce_budget + plan.run(ib, prices={"SOXL": 25.}) + attempt = next(iter(plan.session().read_state()["attempts"].values())) + assert attempt["new_risk"] is True + assert attempt["quantity"] == 3 + assert [order["quantity"] for order in ib.orders] == [3] diff --git a/tests/test_current_risk_fact_precedence.py b/tests/test_current_risk_fact_precedence.py new file mode 100644 index 00000000..ff76f4b3 --- /dev/null +++ b/tests/test_current_risk_fact_precedence.py @@ -0,0 +1,71 @@ +"""Current unsafe account facts cannot be cleared by strategy projections.""" +import pytest + +from application import account_new_risk_gate_support as gate +from tests.test_persistent_drawdown_breaker import session_at + + +@pytest.fixture(autouse=True) +def offline_context(monkeypatch): + monkeypatch.setenv("PORTFOLIO_LOSS_STOPLOSS_ENABLED", "false") + 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) + + +@pytest.mark.parametrize("native,projected", [ + ({"unknown_pending_orders": True}, {"unknown_pending_orders": False}), + ({"unknown_pending_orders": None}, {"unknown_pending_orders": False}), + ({"unknown_pending_orders": "false"}, {"unknown_pending_orders": False}), + ({"production_drift_status": "critical"}, {"production_drift_status": "healthy"}), + ({"daily_loss_usd": 100}, {"daily_loss_usd": 0, "max_daily_loss_usd": 100}), + ({"daily_loss_usd": None}, {"daily_loss_usd": 0, "max_daily_loss_usd": 100}), + ({"realized_vol": float("nan")}, {"realized_vol": .1}), + ({"realized_vol": None}, {"realized_vol": .1}), + ({"realized_vol": "unknown"}, {"realized_vol": .1}), + ({"account_equity_usd": None}, {"equity_usd": 40_000}), + ({"account_equity_usd": float("nan")}, {"equity_usd": 40_000}), + ({"observation_status": "STALE"}, {"observation_status": "COMPLETE"}), + ({"observation_status": "UNKNOWN"}, {"observation_status": "COMPLETE"}), + ({"reconciliation_status": "UNVERIFIED"}, {"reconciliation_status": "VERIFIED"}), + ({"durable_circuit_breaker_state": "UNKNOWN"}, {"durable_circuit_breaker_state": "CLOSED"}), +]) +def test_current_unsafe_fact_blocks_healthy_projection(native, projected, tmp_path): + # Include every healthy projected axis: setdefault must not hide native failures. + metadata = {"account_new_risk_snapshot": { + "equity_usd": 40_000, "observation_status": "COMPLETE", + "reconciliation_status": "VERIFIED", "circuit_breaker_state": "CLOSED", + **projected, + }} + result = gate.evaluate_persistent_account_new_risk_admission( + {"account_equity_usd": 40_000, "equity": 40_000, **native}, + signal_metadata=metadata, session=session_at(tmp_path / "journal"), + ) + assert gate.new_risk_buy_prohibited(result) + assert result.reason_codes + assert result.live_authority_granted is False + assert gate.scale_targets_for_admission({"TQQQ": .025}, result) == {"TQQQ": .025} + + +@pytest.mark.parametrize("native", [ + {"unknown_pending_orders": True}, + {"realized_vol": float("nan")}, + {"observation_status": "STALE"}, + {"total_equity": None}, +]) +def test_direct_portfolio_path_uses_current_facts(native): + result = gate.evaluate_portfolio_new_risk_admission({ + "total_equity": 40_000, **native, + "account_new_risk_snapshot": {"equity_usd": 40_000, "realized_vol": .1, + "unknown_pending_orders": False, "observation_status": "COMPLETE", + "reconciliation_status": "VERIFIED", "circuit_breaker_state": "CLOSED"}, + }) + assert gate.new_risk_buy_prohibited(result) + + +def test_missing_current_equity_cannot_be_supplied_by_old_signal(tmp_path): + result = gate.evaluate_persistent_account_new_risk_admission( + {}, signal_metadata={"account_new_risk_snapshot": {"equity_usd": 40_000}}, + session=session_at(tmp_path / "journal"), + ) + assert gate.new_risk_buy_prohibited(result) diff --git a/tests/test_daily_loss_snapshot_transport.py b/tests/test_daily_loss_snapshot_transport.py new file mode 100644 index 00000000..e42512b7 --- /dev/null +++ b/tests/test_daily_loss_snapshot_transport.py @@ -0,0 +1,36 @@ +"""Native daily-loss transport regressions; fake snapshots and local journals.""" +from dataclasses import replace + +import pytest + +from application import account_new_risk_gate_support as gate +from application.rebalance_service import _snapshot_to_portfolio_view +from tests.test_snapshot_risk_fact_transport import native_snapshot, healthy_projection +from tests.test_persistent_drawdown_breaker import session_at + + +@pytest.fixture(autouse=True) +def offline_context(monkeypatch): + for name in ("IBKR_MAX_DAILY_LOSS_USD", "MAX_DAILY_LOSS_USD", "RUNTIME_TARGET_JSON"): + monkeypatch.delenv(name, raising=False) + monkeypatch.setenv("PORTFOLIO_LOSS_STOPLOSS_ENABLED", "false") + monkeypatch.setattr(gate, "resolve_production_drift_status_from_store", lambda **_kw: None) + + +@pytest.mark.parametrize("daily_loss", [100, 101, None, "invalid", float("nan")]) +def test_native_daily_loss_breach_cannot_be_hidden_by_old_projection(tmp_path, daily_loss): + snapshot = native_snapshot(40000) + snapshot = replace(snapshot, metadata={**snapshot.metadata, "daily_loss_usd": daily_loss}) + _, values = _snapshot_to_portfolio_view(snapshot) + metadata = healthy_projection() + metadata["account_new_risk_snapshot"].update(max_daily_loss_usd=100, daily_loss_usd=0) + # Control: the downstream adapter correctly blocks when it receives the fact. + direct = gate.evaluate_persistent_account_new_risk_admission( + {**values, "daily_loss_usd": daily_loss}, signal_metadata=metadata, + session=session_at(tmp_path / "direct"), + ) + assert gate.new_risk_buy_prohibited(direct) + result = gate.evaluate_persistent_account_new_risk_admission( + values, signal_metadata=metadata, session=session_at(tmp_path / "native"), + ) + assert gate.new_risk_buy_prohibited(result), "native daily loss was discarded and old zero loss admitted" diff --git a/tests/test_drawdown_exit_sizing.py b/tests/test_drawdown_exit_sizing.py new file mode 100644 index 00000000..befb8663 --- /dev/null +++ b/tests/test_drawdown_exit_sizing.py @@ -0,0 +1,151 @@ +"""Synthetic drawdown sizing through real planning and journal dispatch.""" +from types import SimpleNamespace as NS + +import pytest + +from application import account_new_risk_gate_support as gate +from application import execution_service +from quant_platform_kit.common.models import ExecutionReport, OrderIntent +from notifications.telegram import build_translator +from tests.test_execution_service import _signal_metadata, execute_rebalance +from tests.test_persistent_drawdown_breaker import preview, session_at + +BOUNDARIES = [0.074999, 0.075, 0.075001, 0.10, 0.15, 0.20] +PROFILES = [("TQQQ", "tqqq_growth_income"), ("SOXL", "soxl_soxx_trend_income")] + + +@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.setenv("ACCOUNT_NEW_RISK_GATE", "1") + monkeypatch.setenv("PORTFOLIO_LOSS_STOPLOSS_ENABLED", "false") + monkeypatch.setenv("IBKR_DRY_RUN_ONLY", "true") + monkeypatch.setenv("RUNTIME_TARGET_ENABLED", "false") + monkeypatch.setattr(gate, "resolve_production_drift_status_from_store", lambda **_kw: None) + monkeypatch.setattr(execution_service, "maybe_publish_attention_for_admission", lambda *_a, **_kw: {}) + gate.set_cycle_snapshot(None) + yield + gate.set_cycle_snapshot(None) + + +def values_for(drawdown, *, equity=40_000, realized_vol=None): + values = {"equity": equity, "account_equity_usd": equity, "buying_power": equity - 1_000, + "drawdown_from_peak": drawdown, "peak_equity_usd": equity / (1 - drawdown)} + if realized_vol is not None: + values["realized_vol"] = realized_vol + return values + + +def dispatch_cycle(tmp_path, values, *, symbol="TQQQ", target=.025, snapshot=None): + calls = [] + + def fake_venue(_ib, order): + calls.append((order.side, order.quantity)) + return ExecutionReport(symbol=order.symbol, side=order.side, quantity=order.quantity, + status="Filled", broker_order_id="synthetic", + filled_quantity=order.quantity, average_fill_price=100) + + ib = NS(openTrades=lambda: [], reqAllOpenOrders=lambda: [], fills=lambda: [], + accountValues=lambda: [NS(tag="CashBalance", currency="USD", value="39000")]) + metadata = _signal_metadata({symbol: target}, risk_symbols=(symbol,)) + if snapshot is not None: + metadata["account_new_risk_snapshot"].update(snapshot) + _, summary = execute_rebalance( + ib, {symbol: target}, {symbol: {"quantity": 10, "avg_cost": 100}}, values, + fetch_quote_snapshots=lambda _ib, symbols: {s: NS(last_price=100) for s in symbols}, + submit_order_intent=fake_venue, order_intent_cls=OrderIntent, translator=build_translator("en"), + signal_metadata=metadata, dry_run_only=False, execution_mode="paper", + cash_reserve_ratio=0, rebalance_threshold_ratio=0, limit_buy_premium=1, + sell_settle_delay_sec=0, min_order_notional=0, execution_lock_dir=tmp_path, + return_summary=True, acquire_execution_claim=lambda: True, + ) + return calls, summary + + +@pytest.mark.parametrize("symbol,profile", PROFILES) +@pytest.mark.parametrize("drawdown", BOUNDARIES) +def test_switch_off_boundary_keeps_matching_holding(symbol, profile, drawdown, tmp_path): + summary = preview(symbol, profile, {symbol: {"quantity": 10, "avg_cost": 100}}, + values_for(drawdown), {symbol: .025}, tmp_path / "cycle", + session_at(tmp_path / "journal", profile)) + assert summary["orders_submitted"] == [] + assert summary["orders_pending"] == [] + assert {row["symbol"]: row["target_weight"] for row in summary["target_vs_current"]} == {symbol: .025} + + +@pytest.mark.parametrize("symbol", ["TQQQ", "SOXL"]) +@pytest.mark.parametrize("drawdown", BOUNDARIES) +def test_switch_off_fake_dispatch_receives_no_sell(symbol, drawdown, tmp_path): + calls, summary = dispatch_cycle(tmp_path, values_for(drawdown), symbol=symbol) + assert calls == [] + assert summary["orders_submitted"] == [] + assert summary["orders_pending"] == [] + + +@pytest.mark.parametrize("drawdown,quantity", [(d, 0 if d < .075 else 5 if d < .15 else 10) for d in BOUNDARIES]) +def test_switch_on_fake_dispatch_retains_original_sizing(drawdown, quantity, tmp_path, monkeypatch): + monkeypatch.setenv("PORTFOLIO_LOSS_STOPLOSS_ENABLED", "true") + calls, summary = dispatch_cycle(tmp_path, values_for(drawdown)) + assert calls == ([("sell", quantity)] if quantity else []) + assert [row["quantity"] for row in summary["orders_filled"]] == ([quantity] if quantity else []) + + +@pytest.mark.parametrize("equity,vol,scale", [(40_000, .4, .5), (100_000, None, .85), (100_000, .4, .425)]) +@pytest.mark.parametrize("drawdown", [0, .10, .20]) +def test_switch_off_keeps_independent_capital_and_volatility_sizing(equity, vol, scale, drawdown, tmp_path): + target = 1_000 / equity + calls, summary = dispatch_cycle(tmp_path, values_for(drawdown, equity=equity, realized_vol=vol), target=target) + assert summary["target_vs_current"][0]["target_weight"] == pytest.approx(target * scale) + assert calls == [("sell", int(10 * (1 - scale)))] + + +@pytest.mark.parametrize("switch", [None, "", "unknown", "true false", "2"]) +@pytest.mark.parametrize("drawdown", [.10, .20]) +def test_ambiguous_switch_confers_no_drawdown_exit_authority(switch, drawdown, tmp_path, monkeypatch): + if switch is None: + monkeypatch.delenv("PORTFOLIO_LOSS_STOPLOSS_ENABLED", raising=False) + else: + monkeypatch.setenv("PORTFOLIO_LOSS_STOPLOSS_ENABLED", switch) + calls, summary = dispatch_cycle(tmp_path, values_for(drawdown)) + assert calls == [] + assert summary["orders_submitted"] == [] + if drawdown >= .15: + assert gate.new_risk_buy_prohibited(gate.evaluate_account_values_new_risk_admission(values_for(drawdown))) + + +def test_unreadable_switch_confers_no_exit_authority(tmp_path, monkeypatch): + original_get = gate.os.environ.get + + def unreadable(key, default=None): + if key == "PORTFOLIO_LOSS_STOPLOSS_ENABLED": + raise OSError("synthetic switch read failure") + return original_get(key, default) + + monkeypatch.setattr(gate.os.environ, "get", unreadable) + calls, summary = dispatch_cycle(tmp_path, values_for(.10)) + assert calls == [] + assert summary["orders_submitted"] == [] + + +def test_native_cycle_drawdown_does_not_shrink_a_stale_healthy_projection(tmp_path): + calls, summary = dispatch_cycle(tmp_path, values_for(.10), snapshot={ + "equity_usd": 100_000, "peak_equity_usd": 100_000, "drawdown_from_peak": 0, + }) + assert calls == [] + assert summary["target_vs_current"][0]["target_weight"] == .025 + + +def test_restarted_open_keeps_independent_volatility_sizing(tmp_path): + journal = tmp_path / "journal" + original = session_at(journal) + gate.evaluate_persistent_account_new_risk_admission(values_for(.20), session=original) + restarted = session_at(journal, "other-synthetic") + assert restarted.read_state()["circuit_breaker_open"] is True + summary = preview("TQQQ", "tqqq_growth_income", {"TQQQ": {"quantity": 10, "avg_cost": 100}}, + values_for(.10, realized_vol=.4), {"TQQQ": .025}, tmp_path / "cycle", restarted) + assert summary["target_vs_current"][0]["target_weight"] == .0125 + assert summary["orders_submitted"] == [{"symbol": "TQQQ", "side": "sell", "quantity": 5, "status": "dry_run"}] + assert gate.new_risk_buy_prohibited(gate.evaluate_persistent_account_new_risk_admission( + values_for(.10, realized_vol=.4), session=restarted)) + assert session_at(journal).read_state()["circuit_breaker_open"] is True diff --git a/tests/test_native_risk_fact_precedence.py b/tests/test_native_risk_fact_precedence.py new file mode 100644 index 00000000..7a055d41 --- /dev/null +++ b/tests/test_native_risk_fact_precedence.py @@ -0,0 +1,69 @@ +"""Current account evidence must not be hidden by old strategy projections.""" +from copy import deepcopy + +import pytest + +from application import account_new_risk_gate_support as gate +from tests.test_persistent_drawdown_breaker import session_at + + +@pytest.fixture(autouse=True) +def isolated_risk_context(monkeypatch): + monkeypatch.setenv("PORTFOLIO_LOSS_STOPLOSS_ENABLED", "false") + for name in ("IBKR_MAX_DAILY_LOSS_USD", "MAX_DAILY_LOSS_USD", "RUNTIME_TARGET_JSON"): + monkeypatch.delenv(name, raising=False) + monkeypatch.setattr(gate, "resolve_production_drift_status_from_store", lambda **_kw: None) + + +HEALTHY = { + "equity_usd": 100_000, "peak_equity_usd": 100_000, + "drawdown_from_peak": 0, "realized_vol": .1, + "unknown_pending_orders": False, "observation_status": "COMPLETE", + "reconciliation_status": "VERIFIED", "circuit_breaker_state": "CLOSED", +} + + +@pytest.mark.parametrize("native", [ + {"unknown_pending_orders": True}, + {"unknown_pending_orders": None}, + {"unknown_pending_orders": "unknown"}, + {"realized_vol": float("nan")}, + {"realized_vol": None}, + {"account_equity_usd": float("nan"), "equity": 100_000}, + {"peak_equity_usd": None}, + {"observation_status": "STALE"}, + {"observation_status": "UNKNOWN"}, + {"reconciliation_status": "UNVERIFIED"}, +]) +def test_current_unsafe_facts_block_old_healthy_projection(native, tmp_path): + values = {"equity": 100_000, **native} + metadata = {"account_new_risk_snapshot": deepcopy(HEALTHY)} + result = gate.evaluate_persistent_account_new_risk_admission( + values, signal_metadata=metadata, session=session_at(tmp_path), + ) + assert gate.new_risk_buy_prohibited(result) + assert metadata["account_new_risk_snapshot"] == HEALTHY + assert gate.scale_targets_for_admission({"TQQQ": .1}, result) == {"TQQQ": .1} + + +def test_missing_current_equity_does_not_reuse_projection(tmp_path): + result = gate.evaluate_persistent_account_new_risk_admission( + {}, signal_metadata={"account_new_risk_snapshot": deepcopy(HEALTHY)}, + session=session_at(tmp_path), + ) + assert gate.new_risk_buy_prohibited(result) + + +def test_later_native_failure_is_not_hidden_by_prior_cycle_or_restart(tmp_path): + metadata = {"account_new_risk_snapshot": deepcopy(HEALTHY)} + result = gate.evaluate_persistent_account_new_risk_admission( + {"equity": 100_000, "realized_vol": .1}, signal_metadata=metadata, + session=session_at(tmp_path), + ) + assert not gate.new_risk_buy_prohibited(result) + for owner in ("same-process", "restarted-process"): + result = gate.evaluate_persistent_account_new_risk_admission( + {"equity": 100_000, "realized_vol": float("nan")}, + signal_metadata=metadata, session=session_at(tmp_path, owner), + ) + assert gate.new_risk_buy_prohibited(result) diff --git a/tests/test_notif_consolidation_equivalence.py b/tests/test_notif_consolidation_equivalence.py index 1c282c3a..8f807d11 100644 --- a/tests/test_notif_consolidation_equivalence.py +++ b/tests/test_notif_consolidation_equivalence.py @@ -233,6 +233,8 @@ def test_flags_off_entrypoints_match_main_call_recorders(monkeypatch, kind, case for call in expected.get("calls", []): if call[0] == "execute": call[-1]["account_execution_session"] = None + # This legacy fixture has strategy equity but no qualified account NAV. + call[3]["account_equity_usd"] = None if routine: if "calls" in expected: expected["calls"] = [call for call in expected["calls"] if call[0] != "send"] @@ -243,5 +245,6 @@ def test_flags_off_entrypoints_match_main_call_recorders(monkeypatch, kind, case for call in expected.get("calls", []): if call[0] == "execute": call[-1]["account_execution_session"] = None + call[3]["account_equity_usd"] = None assert observed == expected diff --git a/tests/test_persistent_drawdown_breaker.py b/tests/test_persistent_drawdown_breaker.py new file mode 100644 index 00000000..9dcec278 --- /dev/null +++ b/tests/test_persistent_drawdown_breaker.py @@ -0,0 +1,207 @@ +"""Synthetic RC-01C admission boundary; no broker or network access.""" +from datetime import datetime, timezone +from types import SimpleNamespace as NS + +import pytest + +from application import account_new_risk_gate_support as gate +from application import execution_service +from application.rebalance_service import _snapshot_to_portfolio_view, _strategy_portfolio_view +from quant_platform_kit.common.execution_state import AccountSubmissionSession, ExecutionMarkerStore +from quant_platform_kit.common.models import OrderIntent +from tests.test_execution_service import _signal_metadata, execute_rebalance, translate + +PROFILES = [("TQQQ", "tqqq_growth_income"), ("SOXL", "soxl_soxx_trend_income")] + + +@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.setenv("PORTFOLIO_LOSS_STOPLOSS_ENABLED", "false") + monkeypatch.setenv("IBKR_DRY_RUN_ONLY", "true") + monkeypatch.setenv("RUNTIME_TARGET_ENABLED", "false") + monkeypatch.setattr(gate, "resolve_production_drift_status_from_store", lambda **_kw: None) + monkeypatch.setattr(execution_service, "maybe_publish_attention_for_admission", lambda *_a, **_k: {}) + gate.set_cycle_snapshot(None) + yield + gate.set_cycle_snapshot(None) + + +def session_at(path, profile="synthetic", account="DU00000001"): + return AccountSubmissionSession( + ExecutionMarkerStore(local_dir=path), broker="ibkr", account_id=account, + owner_id=profile, profile=profile, intent_id="1" * 64, release_id="2" * 64, + ) + + +def snapshot_for(symbol): + return NS( + total_equity=40_000, buying_power=39_000, + as_of=datetime(2026, 11, 2, tzinfo=timezone.utc), + positions=(NS(symbol=symbol, quantity=10, average_cost=100),), + metadata={"peak_equity_usd": 50_000, "drawdown_from_peak": .2, + "broker_net_liquidation": 40_000, "total_equity_source": "broker_net_liquidation"}, + ) + + +def preview(symbol, profile, positions, values, targets, path, session): + metadata = _signal_metadata( + targets, risk_symbols=tuple(targets), trade_date="2026-11-02", + effective_date=datetime.now(timezone.utc).date().isoformat(), + ) + calls = [] + ib = NS(openTrades=lambda: [], reqAllOpenOrders=lambda: [], fills=lambda: [], + accountValues=lambda: [NS(tag="CashBalance", currency="USD", value="39000")]) + _, summary = execute_rebalance( + ib, targets, positions, values, + fetch_quote_snapshots=lambda _ib, symbols: {s: NS(last_price=100) for s in symbols}, + submit_order_intent=lambda *_a, **_k: calls.append(True), + order_intent_cls=OrderIntent, translator=translate, strategy_profile=profile, + signal_metadata=metadata, dry_run_only=True, cash_reserve_ratio=0, + rebalance_threshold_ratio=0, limit_buy_premium=1, sell_settle_delay_sec=0, + min_order_notional=0, execution_lock_dir=path, return_summary=True, + account_execution_session=session, + ) + assert calls == [] + assert metadata["allocation"]["targets"] == targets + return summary + + +@pytest.mark.parametrize("symbol,profile", PROFILES) +def test_drawdown_survives_native_and_strategy_projection(symbol, profile): + positions, values = _snapshot_to_portfolio_view(snapshot_for(symbol)) + # Strategy allocation equity changes; account drawdown uses whole-account NAV. + positions["VOO"] = {"quantity": 10, "avg_cost": 500} + values["buying_power"] = 10_000 + _, scoped = _strategy_portfolio_view(positions, values, (symbol,)) + assert scoped["equity"] == 11_000 + assert scoped["account_equity_usd"] == 40_000 + observed = gate.build_snapshot_from_portfolio(gate.build_portfolio_from_account_values(scoped)) + assert observed.equity_usd == 40_000 + assert observed.peak_equity_usd == 50_000 + assert observed.drawdown_from_peak == .2 + assert gate.evaluate_account_values_new_risk_admission(scoped).hard_brake_triggered + + +@pytest.mark.parametrize("symbol,profile", PROFILES) +@pytest.mark.parametrize("holding,target", [(10, .025), (10, .05), (0, .025)]) +def test_switch_off_preserves_target_and_blocks_new_positions_and_adds(symbol, profile, holding, target, tmp_path): + positions, values = _snapshot_to_portfolio_view(snapshot_for(symbol)) + positions[symbol]["quantity"] = holding + session = session_at(tmp_path / "journal", profile) + summary = preview(symbol, profile, positions, values, {symbol: target}, tmp_path / "cycle", session) + assert summary["orders_submitted"] == [] + assert summary["orders_pending"] == [] + assert session.read_state()["circuit_breaker_open"] is True + # Admission cannot scale the holding target down, even though its scale is zero. + admission = gate.evaluate_persistent_account_new_risk_admission(values, session=session) + assert admission.combined_scale == 0 + assert gate.scale_targets_for_admission({symbol: target}, admission) == {symbol: target} + assert gate.new_risk_buy_prohibited(admission) + + +@pytest.mark.parametrize("symbol,profile", PROFILES) +def test_recovered_cycles_and_closed_injection_cannot_clear_open(symbol, profile, tmp_path): + positions, values = _snapshot_to_portfolio_view(snapshot_for(symbol)) + session = session_at(tmp_path / "journal", profile) + preview(symbol, profile, positions, values, {symbol: .025}, tmp_path / "trip", session) + for cycle in (1, 2): + recovered = {"equity": 40_000, "buying_power": 39_000, "drawdown_from_peak": 0, + "durable_circuit_breaker_state": "CLOSED"} + restarted = session_at(tmp_path / "journal", "other-synthetic") + admission = gate.evaluate_persistent_account_new_risk_admission( + recovered, session=restarted, signal_metadata={"account_new_risk_snapshot": { + "circuit_breaker_state": "CLOSED", "durable_circuit_breaker_state": "CLOSED", + "reset_authorized": True, + }}, + ) + assert gate.new_risk_buy_prohibited(admission) + assert "CIRCUIT_BREAKER_OPEN" in admission.reason_codes + assert admission.circuit_breaker_reset is False + summary = preview(symbol, profile, positions, recovered, {symbol: .05}, tmp_path / str(cycle), restarted) + assert summary["orders_submitted"] == [] + assert session.read_state()["generation"] == 1 + + +@pytest.mark.parametrize("symbol,profile", PROFILES) +def test_switch_on_preserves_existing_reductions(symbol, profile, tmp_path, monkeypatch): + monkeypatch.setenv("PORTFOLIO_LOSS_STOPLOSS_ENABLED", "true") + positions, values = _snapshot_to_portfolio_view(snapshot_for(symbol)) + summary = preview(symbol, profile, positions, values, {symbol: .025}, tmp_path / "cycle", session_at(tmp_path / "journal", profile)) + assert summary["orders_submitted"] == [{"symbol": symbol, "side": "sell", "quantity": 10, "status": "dry_run"}] + + +def test_durable_open_dominates_closed_projection(): + portfolio = {"total_equity": 40_000, "durable_circuit_breaker_state": "OPEN", + "account_new_risk_snapshot": {"circuit_breaker_state": "CLOSED", "durable_circuit_breaker_state": "CLOSED"}} + assert gate.build_account_new_risk_snapshot(portfolio)["circuit_breaker_state"] == "OPEN" + + +@pytest.mark.parametrize("failure", ["missing", "old_capability", "read", "write", "corruption"]) +def test_persistence_failures_prohibit_new_risk(failure, tmp_path, monkeypatch): + session = session_at(tmp_path) + if failure == "missing": + session = None + elif failure == "old_capability": + monkeypatch.delattr(AccountSubmissionSession, "record_circuit_breaker_open") + elif failure in {"read", "write"}: + method = "read_state" if failure == "read" else "_append" + 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._location(1).write_text("invalid json") + result = gate.evaluate_persistent_account_new_risk_admission({"equity": 40_000, "drawdown_from_peak": .2}, session=session) + assert gate.new_risk_buy_prohibited(result) + assert "PERSISTENT_BREAKER_UNAVAILABLE" in result.reason_codes + + +def test_no_target_cycle_records_open_before_return_and_never_calls_exit(tmp_path): + from application.rebalance_service import run_strategy_core + from application.runtime_dependencies import IBKRRebalanceConfig, IBKRRebalanceRuntime + from quant_platform_kit.common.port_adapters import CallableNotificationPort + # Use the actual runtime ports, a local fake account journal and no broker adapter. + snapshot = snapshot_for("TQQQ") + config = IBKRRebalanceConfig( + translator=lambda key, **_kw: key, separator="---", strategy_profile="synthetic", + account_ids=("DU00000001",), execution_state_store=ExecutionMarkerStore(local_dir=tmp_path / "journal"), + reconciliation_output_path=tmp_path / "reconciliation.json", notify_no_trade_cycles=False, + runtime_release_receipt={"attestation_state": "self_attested", "release_id": "synthetic", "strategy_release": {"revision": "synthetic"}}, + ) + calls = [] + ib = NS(isConnected=lambda: True, disconnect=lambda: None) + runtime = IBKRRebalanceRuntime( + connect_ib=lambda: ib, portfolio_port_factory=lambda _ib: NS(get_portfolio_snapshot=lambda: snapshot), + compute_signals=lambda *_a: (None, "", False, "", {"strategy_profile": "synthetic"}), + execute_rebalance=lambda *_a, **_k: calls.append("execute"), + notifications=CallableNotificationPort(lambda _text: None), + ) + result = run_strategy_core(runtime=runtime, config=config, account_brake_submit=lambda *_a, **_k: calls.append("exit"), + account_brake_session=session_at(tmp_path / "journal"), account_brake_order_intent_cls=OrderIntent) + assert result.target_weights is None + assert result.signal_metadata["account_brake_observation"]["hard_brake_triggered"] is True + assert calls == [] + assert session_at(tmp_path / "journal", "other-synthetic").read_state()["circuit_breaker_open"] is True + + +def test_two_ten_share_holdings_never_become_drawdown_sell_plans(tmp_path): + positions = {symbol: {"quantity": 10, "avg_cost": 100} for symbol in ("TQQQ", "SOXL")} + values = {"equity": 40_000, "buying_power": 38_000, "drawdown_from_peak": .2, "peak_equity_usd": 50_000} + targets = {"TQQQ": .025, "SOXL": .025} + summary = preview("TQQQ", "tqqq_growth_income", positions, values, targets, + tmp_path / "cycle", session_at(tmp_path / "journal")) + assert summary["orders_submitted"] == [] + assert summary["orders_pending"] == [] + assert {row["symbol"]: row["target_weight"] for row in summary["target_vs_current"]} == targets + + +def test_native_drawdown_facts_outrank_a_stale_healthy_signal(): + _, values = _snapshot_to_portfolio_view(snapshot_for("TQQQ")) + metadata = {"account_new_risk_snapshot": {"equity_usd": 100_000, "peak_equity_usd": 100_000, "drawdown_from_peak": 0}} + result = gate.evaluate_account_values_new_risk_admission(values, signal_metadata=metadata) + assert result.hard_brake_triggered + assert gate.new_risk_buy_prohibited(result) + values.pop("drawdown_from_peak") + assert gate.evaluate_account_values_new_risk_admission(values, signal_metadata=metadata).hard_brake_triggered diff --git a/tests/test_rebalance_service.py b/tests/test_rebalance_service.py index 323b5a0f..e44a9dd8 100644 --- a/tests/test_rebalance_service.py +++ b/tests/test_rebalance_service.py @@ -31,6 +31,31 @@ from notifications.telegram import build_translator +def _run_with_synthetic_account_nlv(*, account_nlv, **kwargs): + """Exercise runtime ports with explicitly qualified synthetic account NAV.""" + from application.rebalance_service import _legacy_portfolio_snapshot + from application.runtime_dependencies import IBKRRebalanceRuntime + from quant_platform_kit.common.port_adapters import CallableNotificationPort, CallablePortfolioPort + + read_portfolio = kwargs.pop("get_current_portfolio") + + def read_snapshot(ib): + snapshot = _legacy_portfolio_snapshot(ib, get_current_portfolio=read_portfolio) + return replace(snapshot, metadata={ + "broker_net_liquidation": account_nlv, + "total_equity_source": "broker_net_liquidation", + }) + + runtime = IBKRRebalanceRuntime( + connect_ib=kwargs.pop("connect_ib"), + portfolio_port_factory=lambda ib: CallablePortfolioPort(lambda: read_snapshot(ib)), + compute_signals=kwargs.pop("compute_signals"), + execute_rebalance=kwargs.pop("execute_rebalance"), + notifications=CallableNotificationPort(kwargs.pop("send_tg_message")), + ) + return run_strategy_core(runtime=runtime, **kwargs) + + def test_heartbeat_uses_broker_total_without_changing_strategy_equity(): from datetime import datetime, timezone @@ -956,7 +981,7 @@ def disconnect(self): output_root = tmp_path / "tech_communication_pullback_enhancement" / "reconciliation" - result = run_strategy_core( + result = _run_with_synthetic_account_nlv(account_nlv=800.0, connect_ib=lambda: FakeIB(), get_current_portfolio=lambda _ib: ({}, {"equity": 800.0, "buying_power": 500.0}), compute_signals=lambda _ib, _holdings: ( @@ -1152,7 +1177,7 @@ def execute_with_observed_claim(*args, **kwargs): monkeypatch.setattr(strategy_module, "build_broker_adapters", lambda **_kwargs: state.adapters) def run(**overrides): - return run_strategy_core( + 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}), compute_signals=lambda *_args: (state.target_weights, "synthetic", False, "", state.metadata), @@ -1430,6 +1455,7 @@ def submit(ib, intent, **kwargs): def test_no_target_brake_through_real_session_and_guarded_wrapper(tmp_path, monkeypatch, strategy_module_factory): + monkeypatch.setenv("PORTFOLIO_LOSS_STOPLOSS_ENABLED", "true") import pytest platform = strategy_module_factory() from types import SimpleNamespace @@ -1460,7 +1486,7 @@ def isConnected(self): return True def disconnect(self): return None messages = [] monkeypatch.delenv("IBKR_UNIFIED_NOTIFICATION_ENABLED", raising=False) - result = run_strategy_core(config=config, connect_ib=lambda: FakeIB(), get_current_portfolio=lambda _ib: ({"VOO": {"quantity": 2}}, {"equity": 8000, "buying_power": 0}), compute_signals=lambda _ib, _holdings: (None, "signal", False, "no target", {"strategy_profile": "synthetic", "peak_equity_usd": 9500, "realized_vol": float("nan")}), execute_rebalance=lambda *_a, **_kw: [], send_tg_message=messages.append, translator=_build_test_translator(), separator="---", reconciliation_output_path=tmp_path / "reconciliation", account_brake_session=session, account_brake_submit=platform.submit_market_order_intent, account_brake_order_intent_cls=OrderIntent) + result = _run_with_synthetic_account_nlv(account_nlv=8000.0, config=config, connect_ib=lambda: FakeIB(), get_current_portfolio=lambda _ib: ({"VOO": {"quantity": 2}}, {"equity": 8000, "buying_power": 0}), compute_signals=lambda _ib, _holdings: (None, "signal", False, "no target", {"strategy_profile": "synthetic", "peak_equity_usd": 9500, "realized_vol": float("nan")}), execute_rebalance=lambda *_a, **_kw: [], send_tg_message=messages.append, translator=_build_test_translator(), separator="---", reconciliation_output_path=tmp_path / "reconciliation", account_brake_session=session, account_brake_submit=platform.submit_market_order_intent, account_brake_order_intent_cls=OrderIntent) assert result.signal_metadata["account_brake_observation"]["hard_brake_triggered"] assert len(submitted) == 1 and submitted[0].side == "sell" and submitted[0].quantity == 2 assert messages and "strategy rebalance decision unavailable" in messages[0] diff --git a/tests/test_settled_cash_intervening_updates.py b/tests/test_settled_cash_intervening_updates.py index 9e91d721..dbc1a6ea 100644 --- a/tests/test_settled_cash_intervening_updates.py +++ b/tests/test_settled_cash_intervening_updates.py @@ -186,8 +186,8 @@ def test_post_reservation_update_blocks_final_write_and_retains_unknown(native, plan = native({'SOXL': 400.}, cash=500.) ib = FakeIB(cash=500., settled=500.) reserve = AccountSubmissionSession.reserve - def reserve_then_update(session, order): - grant = reserve(session, order) + def reserve_then_update(session, order, **kwargs): + grant = reserve(session, order, **kwargs) emit_settlement(ib, latest) return grant monkeypatch.setattr(AccountSubmissionSession, 'reserve', reserve_then_update) diff --git a/tests/test_snapshot_risk_fact_transport.py b/tests/test_snapshot_risk_fact_transport.py new file mode 100644 index 00000000..3e732afa --- /dev/null +++ b/tests/test_snapshot_risk_fact_transport.py @@ -0,0 +1,124 @@ +"""Synthetic native snapshot -> account gate regressions; no broker access.""" +from dataclasses import replace + +import pytest + +from application import account_new_risk_gate_support as gate +from application.ibkr_portfolio import fetch_portfolio_snapshot +from application.rebalance_service import ( + _snapshot_to_portfolio_view, + _strategy_portfolio_view, + run_strategy_core, +) +from application.runtime_dependencies import IBKRRebalanceConfig, IBKRRebalanceRuntime +from quant_platform_kit.common.execution_state import ExecutionMarkerStore +from quant_platform_kit.common.port_adapters import CallableNotificationPort +from tests.test_ibkr_portfolio import ( + SnapshotIB, _stock_mark, _stock_position, _usd_account_values, +) +from tests.test_persistent_drawdown_breaker import session_at +from types import SimpleNamespace as NS + + +@pytest.fixture(autouse=True) +def offline(monkeypatch): + monkeypatch.setenv("PORTFOLIO_LOSS_STOPLOSS_ENABLED", "false") + 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) + + +def native_snapshot(nlv=32000): + account = "DU00000001" + rows = _usd_account_values(account, 1000, nlv) + if nlv is None: + rows = [row for row in rows if row.tag != "NetLiquidation"] + ib = SnapshotIB( + [_stock_position(account, "TQQQ", 1, 390, 100)], + [_stock_mark(account, "TQQQ", 1, 390, 39000)], rows, + ) + return fetch_portfolio_snapshot(ib, account_ids=(account,), wait_seconds=0) + + +def healthy_projection(): + return {"account_new_risk_snapshot": { + "equity_usd": 40000, "peak_equity_usd": 40000, "realized_vol": .1, + "observation_status": "COMPLETE", "reconciliation_status": "VERIFIED", + "circuit_breaker_state": "CLOSED", "production_drift_status": "healthy", + }} + + +def test_native_nlv_survives_strategy_scope_and_restart(tmp_path): + snapshot = native_snapshot() + positions, values = _snapshot_to_portfolio_view(snapshot) + assert values["equity"] == 40000 + assert values["account_equity_usd"] == 32000 + _, scoped = _strategy_portfolio_view(positions, values, ("TQQQ",)) + assert scoped["account_equity_usd"] == 32000 + result = gate.evaluate_persistent_account_new_risk_admission( + scoped, signal_metadata=healthy_projection(), session=session_at(tmp_path), + ) + assert "DRAWDOWN_BRAKE_TRIPPED" in result.reason_codes + assert result.hard_brake_triggered + assert session_at(tmp_path).read_state()["circuit_breaker_open"] is True + recovered = gate.evaluate_persistent_account_new_risk_admission( + {"account_equity_usd": 40000}, signal_metadata=healthy_projection(), + session=session_at(tmp_path, profile="other-synthetic"), + ) + assert "CIRCUIT_BREAKER_OPEN" in recovered.reason_codes + + +@pytest.mark.parametrize("evidence", ["missing", "unqualified", "null", "nan", "zero", "negative", "malformed"]) +def test_unavailable_native_nlv_cannot_fall_back_to_strategy_or_projection(tmp_path, evidence): + snapshot = native_snapshot(None if evidence == "missing" else 40000) + metadata = dict(snapshot.metadata) + if evidence == "unqualified": + metadata["total_equity_source"] = "unverified_net_liquidation" + elif evidence != "missing": + metadata["broker_net_liquidation"] = { + "null": None, "nan": float("nan"), "zero": 0, + "negative": -1, "malformed": "invalid", + }[evidence] + snapshot = replace(snapshot, metadata=metadata) + positions, values = _snapshot_to_portfolio_view(snapshot) + assert values["equity"] == 40000 + _, scoped = _strategy_portfolio_view(positions, values, ("TQQQ",)) + result = gate.evaluate_persistent_account_new_risk_admission( + scoped, signal_metadata=healthy_projection(), session=session_at(tmp_path), + ) + assert gate.new_risk_buy_prohibited(result) + assert "EQUITY_UNKNOWN_FAIL_CLOSED" in result.reason_codes + + +@pytest.mark.parametrize("axis,value,reason", [ + ("realized_vol", float("nan"), "INVALID_REALIZED_VOL_FAIL_CLOSED"), + ("observation_status", "STALE", "OBSERVATION_NOT_COMPLETE"), + ("reconciliation_status", "UNVERIFIED", "RECONCILIATION_NOT_VERIFIED"), + ("circuit_breaker_state", "OPEN", "CIRCUIT_BREAKER_OPEN"), + ("production_drift_status", "critical", "PRODUCTION_DRIFT_CRITICAL"), +]) +def test_native_unsafe_axis_reaches_real_cycle_admission(tmp_path, axis, value, reason): + snapshot = native_snapshot(40000) + snapshot = replace(snapshot, metadata={**snapshot.metadata, axis: value}) + calls = [] + store = ExecutionMarkerStore(local_dir=tmp_path / "journal") + config = IBKRRebalanceConfig( + translator=lambda key, **_kw: key, separator="---", strategy_profile="synthetic", + account_ids=("DU00000001",), execution_state_store=store, + reconciliation_output_path=tmp_path / "reconciliation.json", notify_no_trade_cycles=False, + runtime_release_receipt={"attestation_state": "self_attested", "release_id": "synthetic", + "strategy_release": {"revision": "synthetic"}}, + ) + runtime = IBKRRebalanceRuntime( + connect_ib=lambda: NS(isConnected=lambda: True, disconnect=lambda: None), + portfolio_port_factory=lambda _ib: NS(get_portfolio_snapshot=lambda: snapshot), + compute_signals=lambda *_a: (None, "", False, "", {"strategy_profile": "synthetic", **healthy_projection()}), + execute_rebalance=lambda *_a, **_kw: calls.append("execute"), + notifications=CallableNotificationPort(lambda _text: None), + ) + result = run_strategy_core( + runtime=runtime, config=config, + account_brake_submit=lambda *_a, **_kw: calls.append("exit"), + ) + assert reason in result.signal_metadata["account_brake_observation"]["reason_codes"] + assert calls == [] diff --git a/tests/test_unified_notification_adapter.py b/tests/test_unified_notification_adapter.py index 4d95da9f..f9767964 100644 --- a/tests/test_unified_notification_adapter.py +++ b/tests/test_unified_notification_adapter.py @@ -291,6 +291,7 @@ def keep(data): for call in expected["calls"]: if call[0] == "execute": call[-1]["account_execution_session"] = None + call[3]["account_equity_usd"] = None assert keep(observed) == keep(expected) assert observed["result"] == baseline[f"cycle:{case}:en"]["result"] diff --git a/uv.lock b/uv.lock index 3b73764a..cf5a9873 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=6136b1c667a60d484f500c6d6046c0585c75305d" }] +overrides = [{ name = "quant-platform-kit", git = "https://github.com/QuantStrategyLab/QuantPlatformKit.git?rev=cee53965b3d6b368445193b580aa94f2bbb9537c" }] [[package]] name = "beautifulsoup4" @@ -791,7 +791,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=6136b1c667a60d484f500c6d6046c0585c75305d" }, + { name = "quant-platform-kit", git = "https://github.com/QuantStrategyLab/QuantPlatformKit.git?rev=cee53965b3d6b368445193b580aa94f2bbb9537c" }, { name = "requests" }, { name = "ruff", marker = "extra == 'test'" }, { name = "us-equity-strategies", git = "https://github.com/QuantStrategyLab/UsEquityStrategies.git?rev=3f112f5352efe677a0c480c4ab621b8ea3e85336" }, @@ -1327,7 +1327,7 @@ wheels = [ [[package]] name = "quant-platform-kit" version = "1.0.0" -source = { git = "https://github.com/QuantStrategyLab/QuantPlatformKit.git?rev=6136b1c667a60d484f500c6d6046c0585c75305d#6136b1c667a60d484f500c6d6046c0585c75305d" } +source = { git = "https://github.com/QuantStrategyLab/QuantPlatformKit.git?rev=cee53965b3d6b368445193b580aa94f2bbb9537c#cee53965b3d6b368445193b580aa94f2bbb9537c" } [[package]] name = "requests"