From 8e25ae2065f61338b1d64bcffbcde1cc405024f2 Mon Sep 17 00:00:00 2001 From: Pigbibi <20649888+Pigbibi@users.noreply.github.com> Date: Tue, 4 Aug 2026 17:52:18 +0800 Subject: [PATCH 1/3] fix(risk): enforce exposure and trailing state gates Co-Authored-By: Codex --- pyproject.toml | 2 +- qsl.toml | 2 +- src/crypto_strategies/entrypoints/__init__.py | 95 +++++-- src/crypto_strategies/entrypoints/_common.py | 84 ++++-- .../crypto_live_pool_rotation/rotation.py | 90 +++++++ tests/test_entrypoint_risk_gate.py | 225 +++++++++++++++- tests/test_entrypoints.py | 245 +++++++++++++++++- tests/test_rotation_authority.py | 56 +++- uv.lock | 4 +- 9 files changed, 754 insertions(+), 49 deletions(-) diff --git a/pyproject.toml b/pyproject.toml index 84823aa..1e14bab 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -9,7 +9,7 @@ description = "Shared crypto strategy catalog and implementations" readme = "README.md" requires-python = ">=3.11" dependencies = [ - "quant-platform-kit @ git+https://github.com/QuantStrategyLab/QuantPlatformKit.git@776fe71e57e2924fcd1c73126f41d244242240bb", + "quant-platform-kit @ git+https://github.com/QuantStrategyLab/QuantPlatformKit.git@b371322b948e4298920a7d8613b155245dcd5f8d", ] [tool.setuptools] diff --git a/qsl.toml b/qsl.toml index 63412ea..64c195c 100644 --- a/qsl.toml +++ b/qsl.toml @@ -4,5 +4,5 @@ upgrade_ring = "ring_b" [compat] bundle = "2026.07.4" requires = [ - "quant-platform-kit @ git+https://github.com/QuantStrategyLab/QuantPlatformKit.git@776fe71e57e2924fcd1c73126f41d244242240bb", + "quant-platform-kit @ git+https://github.com/QuantStrategyLab/QuantPlatformKit.git@b371322b948e4298920a7d8613b155245dcd5f8d", ] diff --git a/src/crypto_strategies/entrypoints/__init__.py b/src/crypto_strategies/entrypoints/__init__.py index 9376d9f..68fa383 100644 --- a/src/crypto_strategies/entrypoints/__init__.py +++ b/src/crypto_strategies/entrypoints/__init__.py @@ -2,6 +2,7 @@ from collections.abc import Callable, Mapping from copy import deepcopy +import math from quant_platform_kit.strategy_contracts import ( BudgetIntent, @@ -19,6 +20,10 @@ ) from ._common import apply_risk_gate, record_strategy_decision +from crypto_strategies.strategies.crypto_live_pool_rotation.rotation import ( + build_strategy_stop_evaluation, + evaluate_held_trend_stops, +) """Unified crypto strategy entrypoints built on top of legacy core/rotation modules.""" @@ -116,6 +121,31 @@ def _set_symbol_trade_state(state, symbol, symbol_state): return _get_symbol_trade_state, _set_symbol_trade_state +def _resolve_held_risk_symbols(ctx: StrategyContext, state): + held = { + str(symbol).strip().upper() + for symbol, payload in state.items() + if isinstance(payload, Mapping) + and payload.get("is_holding") + and str(symbol).strip().upper() != "BTCUSDT" + } + snapshot = _resolve_portfolio_snapshot(ctx) + for position in getattr(snapshot, "positions", ()) or (): + symbol = str(getattr(position, "symbol", "")).strip().upper() + quantity = getattr(position, "quantity", 0.0) + market_value = getattr(position, "market_value", 0.0) + values = (quantity, market_value) + if symbol and symbol != "BTCUSDT" and any( + isinstance(value, (int, float)) + and not isinstance(value, bool) + and math.isfinite(float(value)) + and float(value) != 0.0 + for value in values + ): + held.add(symbol) + return tuple(sorted(held)) + + def _load_legacy_modules(): from crypto_strategies.strategies.crypto_live_pool_rotation import core as legacy_core from crypto_strategies.strategies.crypto_live_pool_rotation import rotation as legacy_rotation @@ -159,25 +189,22 @@ def evaluate_crypto_live_pool_rotation(ctx: StrategyContext) -> StrategyDecision weight_mode=str(config.get("weight_mode", "inverse_vol")), ) - sell_reasons: dict[str, str] = {} atr_multiplier = float(config.get("atr_multiplier", 2.5)) - for symbol in trend_universe_symbols: - curr_price = prices.get(symbol) - if curr_price is None: - continue - reason = legacy_rotation.get_trend_sell_reason( - working_state, - symbol, - curr_price, - indicators_map.get(symbol), - selected_candidates, - atr_multiplier, - get_symbol_trade_state_fn=get_symbol_trade_state_fn, - set_symbol_trade_state_fn=set_symbol_trade_state_fn, - translate_fn=translator, - ) - if reason: - sell_reasons[symbol] = str(reason) + held_risk_symbols = _resolve_held_risk_symbols( + ctx, + working_state, + ) + sell_reasons, stop_input_blocked = evaluate_held_trend_stops( + working_state, + held_symbols=held_risk_symbols, + prices=prices, + indicators_map=indicators_map, + selected_candidates=selected_candidates, + atr_multiplier=atr_multiplier, + get_symbol_trade_state_fn=get_symbol_trade_state_fn, + set_symbol_trade_state_fn=set_symbol_trade_state_fn, + translate_fn=translator, + ) eligible_buy_symbols, planned_trend_buys = legacy_rotation.plan_trend_buys( working_state, @@ -200,6 +227,8 @@ def evaluate_crypto_live_pool_rotation(ctx: StrategyContext) -> StrategyDecision ] trend_target_ratio = float(budgets["trend_target_ratio"]) for symbol, payload in sorted(selected_candidates.items()): + if symbol in sell_reasons: + continue positions.append( PositionTarget( symbol=symbol, @@ -257,11 +286,37 @@ def evaluate_crypto_live_pool_rotation(ctx: StrategyContext) -> StrategyDecision } decision = StrategyDecision( positions=tuple(positions), - budgets=budget_intents, - risk_flags=risk_flags, + budgets=() if sell_reasons else budget_intents, + risk_flags=risk_flags + (("rejected:strategy_stop_input",) if stop_input_blocked else ()), diagnostics=diagnostics, ) + if sell_reasons: + decision = StrategyDecision( + positions=(), + budgets=(), + risk_flags=decision.risk_flags, + diagnostics=decision.diagnostics, + ) decision = apply_risk_gate(decision, ctx=ctx) + member_assessment = decision.diagnostics["member_risk_assessment"] + stop_outcome = "TRIGGERED" if sell_reasons else "CLEAR" + stop_action_result = "NOT_REQUIRED" + if stop_outcome == "TRIGGERED": + stop_action_result = "BLOCKED" + decision = StrategyDecision( + positions=decision.positions, + budgets=decision.budgets, + risk_flags=decision.risk_flags, + diagnostics={ + **dict(decision.diagnostics), + "strategy_stop_evaluation": build_strategy_stop_evaluation( + evaluated_at=member_assessment["evaluated_at"], + decision_digest_sha256=member_assessment["decision_digest_sha256"], + outcome=stop_outcome, + action_result=stop_action_result, + ), + }, + ) record_strategy_decision( ctx, decision, diff --git a/src/crypto_strategies/entrypoints/_common.py b/src/crypto_strategies/entrypoints/_common.py index c022612..bbd1732 100644 --- a/src/crypto_strategies/entrypoints/_common.py +++ b/src/crypto_strategies/entrypoints/_common.py @@ -1,13 +1,15 @@ from __future__ import annotations import logging +import math from collections.abc import Mapping +from dataclasses import asdict from typing import Any -from quant_platform_kit.risk.gate import apply_risk_gate as _qpk_apply_risk_gate +from quant_platform_kit.risk.gate import assess_with_evidence as _qpk_assess_with_evidence from quant_platform_kit.risk.gate import enrich_decision_risk_diagnostics from quant_platform_kit.risk.portfolio_diagnostics import extract_portfolio_risk_diagnostics -from quant_platform_kit.strategy_contracts import PositionTarget, StrategyContext, StrategyDecision +from quant_platform_kit.strategy_contracts import StrategyContext, StrategyDecision from quant_platform_kit.strategy_lifecycle.performance_monitor import PerformanceMonitor logger = logging.getLogger(__name__) @@ -49,16 +51,16 @@ def apply_risk_gate( decision: StrategyDecision, *, ctx: StrategyContext | None = None, - max_single_weight: float = 1.0, - max_positions: int = 20, - max_total_exposure: float = 1.0, + max_single_weight: float | None = None, portfolio_snapshot: Any | None = None, market_data: Mapping[str, Any] | None = None, ) -> StrategyDecision: - """QPK unified risk gate: stop-loss, circuit breaker, concentration (task 8).""" - snapshot = portfolio_snapshot if portfolio_snapshot is not None else ( - ctx.portfolio if ctx is not None else None - ) + """Run the QPK MEMBER gate and propagate only its redacted assessment.""" + snapshot = portfolio_snapshot + if snapshot is None and ctx is not None: + snapshot = ctx.portfolio + if snapshot is None: + snapshot = ctx.market_data.get("portfolio_snapshot") if snapshot is not None: portfolio_diag = extract_portfolio_risk_diagnostics(snapshot) decision = enrich_decision_risk_diagnostics( @@ -68,11 +70,63 @@ def apply_risk_gate( ) if market_data is None and ctx is not None: market_data = dict(ctx.market_data or {}) - return _qpk_apply_risk_gate( + mandate_provenance = None if ctx is None else ctx.artifacts.get("mandate_provenance") + if not isinstance(mandate_provenance, Mapping): + mandate_provenance = {} + result = _qpk_assess_with_evidence( decision, - max_single_weight=max_single_weight, - max_positions=max_positions, - max_total_exposure=max_total_exposure, - portfolio_snapshot=snapshot, - market_data=market_data, + snapshot, + scope="MEMBER", + mandate_provenance=mandate_provenance, + market_data=market_data or {}, + ) + risk_flags = tuple( + dict.fromkeys(tuple(decision.risk_flags or ()) + tuple(result.decision.risk_flags or ())) + ) + strategy_weights = [] + strategy_weight_invalid = False + for position in result.decision.positions: + try: + weight = float(position.target_weight) + except (TypeError, ValueError): + strategy_weight_invalid = True + break + if not math.isfinite(weight): + strategy_weight_invalid = True + break + strategy_weights.append(abs(weight)) + strategy_concentration_rejected = False + if max_single_weight is not None: + cap = float(max_single_weight) + if not math.isfinite(cap) or not 0.0 <= cap <= 1.0: + raise ValueError("max_single_weight must be finite and between 0 and 1") + strategy_concentration_rejected = strategy_weight_invalid or any( + weight > cap for weight in strategy_weights + ) + if strategy_concentration_rejected: + risk_flags = tuple(dict.fromkeys(risk_flags + ("rejected:strategy_concentration",))) + strategy_position_count_rejected = len(result.decision.positions) > 20 + if strategy_position_count_rejected: + risk_flags = tuple(dict.fromkeys(risk_flags + ("rejected:too_many_positions",))) + total_exposure = sum(strategy_weights) + strategy_total_exposure_rejected = ( + strategy_weight_invalid + or not math.isfinite(total_exposure) + or total_exposure > 1.0 + 1e-9 + ) + if strategy_total_exposure_rejected: + risk_flags = tuple(dict.fromkeys(risk_flags + ("rejected:overexposed",))) + strategy_rejected = ( + strategy_concentration_rejected + or strategy_position_count_rejected + or strategy_total_exposure_rejected + ) + return StrategyDecision( + positions=() if strategy_rejected else result.decision.positions, + budgets=() if strategy_rejected else result.decision.budgets, + risk_flags=risk_flags, + diagnostics={ + **dict(result.decision.diagnostics or {}), + "member_risk_assessment": asdict(result.assessment), + }, ) diff --git a/src/crypto_strategies/strategies/crypto_live_pool_rotation/rotation.py b/src/crypto_strategies/strategies/crypto_live_pool_rotation/rotation.py index 7c2d422..447a2d8 100644 --- a/src/crypto_strategies/strategies/crypto_live_pool_rotation/rotation.py +++ b/src/crypto_strategies/strategies/crypto_live_pool_rotation/rotation.py @@ -2,7 +2,97 @@ from __future__ import annotations +from collections.abc import Mapping from datetime import datetime, timezone +import math + + +_STRATEGY_STOP_POLICY_ID = "crypto_live_pool_rotation.executable_stop" +_STRATEGY_STOP_POLICY_VERSION = "v1" + + +def _finite_number(value): + if isinstance(value, bool) or not isinstance(value, (int, float)): + return None + number = float(value) + return number if math.isfinite(number) else None + + +def evaluate_held_trend_stops( + state, + *, + held_symbols, + prices, + indicators_map, + selected_candidates, + atr_multiplier, + get_symbol_trade_state_fn, + set_symbol_trade_state_fn, + translate_fn, +): + """Evaluate every held risk symbol; incomplete inputs block CLEAR.""" + sell_reasons = {} + input_blocked = False + for symbol in _normalize_symbol_list(held_symbols): + symbol_state = get_symbol_trade_state_fn(state, symbol) + persisted_symbol_state = state.get(symbol) if isinstance(state, Mapping) else None + indicators = indicators_map.get(symbol) + curr_price = _finite_number(prices.get(symbol)) + atr = _finite_number(indicators.get("atr14")) if isinstance(indicators, Mapping) else None + sma60 = _finite_number(indicators.get("sma60")) if isinstance(indicators, Mapping) else None + entry_price = _finite_number(symbol_state.get("entry_price")) + highest_price = ( + _finite_number(persisted_symbol_state.get("highest_price")) + if isinstance(persisted_symbol_state, Mapping) + and "highest_price" in persisted_symbol_state + else None + ) + if ( + not symbol_state.get("is_holding") + or curr_price is None + or atr is None + or sma60 is None + or entry_price is None + or entry_price <= 0.0 + or highest_price is None + or highest_price <= 0.0 + or highest_price < entry_price + ): + input_blocked = True + sell_reasons[symbol] = translate_fn("trend_sell_reason_missing_stop_input") + continue + reason = get_trend_sell_reason( + state, + symbol, + curr_price, + indicators, + selected_candidates, + atr_multiplier, + get_symbol_trade_state_fn=get_symbol_trade_state_fn, + set_symbol_trade_state_fn=set_symbol_trade_state_fn, + translate_fn=translate_fn, + ) + if reason: + sell_reasons[symbol] = str(reason) + return sell_reasons, input_blocked + + +def build_strategy_stop_evaluation( + *, + evaluated_at, + decision_digest_sha256, + outcome, + action_result, +): + return { + "evaluated": True, + "policy_id": _STRATEGY_STOP_POLICY_ID, + "policy_version": _STRATEGY_STOP_POLICY_VERSION, + "evaluated_at": evaluated_at, + "decision_digest_sha256": decision_digest_sha256, + "outcome": outcome, + "action_result": action_result, + } def _normalize_symbol_list(symbols): diff --git a/tests/test_entrypoint_risk_gate.py b/tests/test_entrypoint_risk_gate.py index 7593219..ea07176 100644 --- a/tests/test_entrypoint_risk_gate.py +++ b/tests/test_entrypoint_risk_gate.py @@ -1,13 +1,36 @@ from __future__ import annotations -from datetime import datetime, timezone +from datetime import datetime, timedelta, timezone +from unittest.mock import patch from quant_platform_kit.common.models import PortfolioSnapshot, Position -from quant_platform_kit.strategy_contracts import PositionTarget, StrategyContext, StrategyDecision +from quant_platform_kit.risk.contracts import RiskGateAssessment, RiskGateResult +from quant_platform_kit.strategy_contracts import BudgetIntent, PositionTarget, StrategyContext, StrategyDecision from crypto_strategies.entrypoints._common import apply_risk_gate +def _zero_cap_mandate(now: datetime) -> dict[str, object]: + return { + "mandate_id": "binance_crypto_research_only_v1", + "mandate_version": "2026-08-04.1", + "authority_receipt_sha256": "246c39b8023b25f913bf1e67dc175005955a7102f3727dfc1bd8e981cf8128ee", + "authority_scope": "RESEARCH_ONLY", + "strategy_profile": "crypto_live_pool_rotation", + "account_mode": "single_strategy_account_v1", + "effective_at": (now - timedelta(minutes=1)).isoformat().replace("+00:00", "Z"), + "expires_at": (now + timedelta(minutes=1)).isoformat().replace("+00:00", "Z"), + "max_snapshot_age_seconds": 300, + "effective_exposure_cap": 0.0, + "loss_budget": 0.0, + "product_caps": 0.0, + "nominal_caps": 0.0, + "product_leverage_factors": {}, + "allowed_nonzero_assets": [], + "source_revision": "b371322b948e4298920a7d8613b155245dcd5f8d", + } + + def test_apply_risk_gate_enriches_stop_loss_diagnostics_from_portfolio() -> None: snapshot = PortfolioSnapshot( as_of=datetime(2026, 7, 9, tzinfo=timezone.utc), @@ -21,4 +44,200 @@ def test_apply_risk_gate_enriches_stop_loss_diagnostics_from_portfolio() -> None decision = StrategyDecision(positions=(PositionTarget(symbol="BTCUSDT", target_weight=0.5),)) result = apply_risk_gate(decision, ctx=ctx) assert result.positions == () - assert "rejected:stop_loss" in result.risk_flags + assert "rejected:risk_gate_assessment" in result.risk_flags + assert result.diagnostics["member_risk_assessment"]["outcome"] == "REJECT" + + +def test_apply_risk_gate_uses_member_evidence_and_zero_cap_clears_authority() -> None: + now = datetime.now(timezone.utc) + snapshot = PortfolioSnapshot( + as_of=now, + total_equity=1000.0, + metadata={ + "observed_effective_exposure": 0.0, + "private_position_rows": [{"symbol": "BTCUSDT", "quantity": 123.0}], + }, + ) + ctx = StrategyContext( + as_of=now, + portfolio=snapshot, + market_data={"private_api_token": "must-not-propagate"}, + artifacts={"mandate_provenance": _zero_cap_mandate(now)}, + ) + decision = StrategyDecision( + positions=(PositionTarget(symbol="BTCUSDT", target_weight=0.1),), + budgets=(BudgetIntent(name="btc", amount=1.0),), + ) + + result = apply_risk_gate(decision, ctx=ctx) + + assessment = result.diagnostics["member_risk_assessment"] + assert result.positions == () + assert result.budgets == () + assert assessment["contract_version"] == "qsl.risk_gate_assessment.v1" + assert assessment["scope"] == "MEMBER" + assert assessment["outcome"] == "REJECT" + assert assessment["mandate_id"] == "binance_crypto_research_only_v1" + assert "private_position_rows" not in repr(assessment) + assert "must-not-propagate" not in repr(assessment) + + +def test_apply_risk_gate_preserves_stricter_strategy_concentration_cap() -> None: + now = datetime.now(timezone.utc) + mandate = _zero_cap_mandate(now) + mandate.update( + { + "mandate_id": "synthetic_algorithm_equivalence_only", + "effective_exposure_cap": 1.0, + "loss_budget": 1000.0, + "product_caps": {"BTCUSDT": 1.0}, + "nominal_caps": {"BTCUSDT": 1.0}, + "product_leverage_factors": {"BTCUSDT": 1}, + "allowed_nonzero_assets": ["BTCUSDT"], + } + ) + snapshot = PortfolioSnapshot( + as_of=now, + total_equity=1000.0, + metadata={"observed_effective_exposure": 0.0}, + ) + ctx = StrategyContext( + as_of=now, + portfolio=snapshot, + artifacts={"mandate_provenance": mandate}, + ) + + result = apply_risk_gate( + StrategyDecision(positions=(PositionTarget(symbol="BTCUSDT", target_weight=0.6),)), + ctx=ctx, + max_single_weight=0.5, + ) + + assert result.diagnostics["member_risk_assessment"]["outcome"] == "APPROVE" + assert result.positions == () + assert result.budgets == () + assert "rejected:strategy_concentration" in result.risk_flags + + +def test_apply_risk_gate_preserves_hard_position_count_limit() -> None: + now = datetime.now(timezone.utc) + symbols = [f"ASSET{index}USDT" for index in range(21)] + mandate = _zero_cap_mandate(now) + mandate.update( + { + "mandate_id": "synthetic_algorithm_equivalence_only", + "effective_exposure_cap": 1.0, + "loss_budget": 1000.0, + "product_caps": {symbol: 1.0 for symbol in symbols}, + "nominal_caps": {symbol: 1.0 for symbol in symbols}, + "product_leverage_factors": {symbol: 1 for symbol in symbols}, + "allowed_nonzero_assets": symbols, + } + ) + snapshot = PortfolioSnapshot( + as_of=now, + total_equity=1000.0, + metadata={"observed_effective_exposure": 0.0}, + ) + ctx = StrategyContext( + as_of=now, + portfolio=snapshot, + artifacts={"mandate_provenance": mandate}, + ) + decision = StrategyDecision( + positions=tuple( + PositionTarget(symbol=symbol, target_weight=0.04) for symbol in symbols + ), + budgets=(BudgetIntent(name="portfolio", amount=1.0),), + ) + + result = apply_risk_gate(decision, ctx=ctx) + + assert result.diagnostics["member_risk_assessment"]["outcome"] == "APPROVE" + assert result.positions == () + assert result.budgets == () + assert "rejected:too_many_positions" in result.risk_flags + + +def test_apply_risk_gate_preserves_hard_total_exposure_limit() -> None: + now = datetime.now(timezone.utc) + symbols = [f"ASSET{index}USDT" for index in range(5)] + mandate = _zero_cap_mandate(now) + mandate.update( + { + "mandate_id": "synthetic_algorithm_equivalence_only", + "effective_exposure_cap": 2.0, + "loss_budget": 1000.0, + "product_caps": {symbol: 1.0 for symbol in symbols}, + "nominal_caps": {symbol: 1.0 for symbol in symbols}, + "product_leverage_factors": {symbol: 1 for symbol in symbols}, + "allowed_nonzero_assets": symbols, + } + ) + snapshot = PortfolioSnapshot( + as_of=now, + total_equity=1000.0, + metadata={"observed_effective_exposure": 0.0}, + ) + ctx = StrategyContext( + as_of=now, + portfolio=snapshot, + artifacts={"mandate_provenance": mandate}, + ) + decision = StrategyDecision( + positions=tuple( + PositionTarget(symbol=symbol, target_weight=0.25) for symbol in symbols + ), + budgets=(BudgetIntent(name="portfolio", amount=1.0),), + ) + + permissive_assessment = RiskGateAssessment( + contract_version="qsl.risk_gate_assessment.v1", + scope="MEMBER", + evaluated_at=now.isoformat().replace("+00:00", "Z"), + policy_id="qpk.risk_gate", + policy_version="v1", + qpk_source_revision="b371322b948e4298920a7d8613b155245dcd5f8d", + mandate_id="synthetic_algorithm_equivalence_only", + mandate_version="test-v1", + mandate_authority_receipt_sha256="a" * 64, + mandate_scope="RESEARCH_ONLY", + decision_digest_sha256="b" * 64, + portfolio_snapshot_digest_sha256="c" * 64, + effective_exposure_cap=2.0, + observed_effective_exposure=0.0, + proposed_effective_exposure=1.25, + outcome="APPROVE", + reason_codes=(), + ) + with patch( + "crypto_strategies.entrypoints._common._qpk_assess_with_evidence", + return_value=RiskGateResult( + decision=decision, + assessment=permissive_assessment, + ), + ): + result = apply_risk_gate(decision, ctx=ctx) + + assert result.diagnostics["member_risk_assessment"]["outcome"] == "APPROVE" + assert result.positions == () + assert result.budgets == () + assert "rejected:overexposed" in result.risk_flags + + for invalid_weight in (None, float("nan"), float("inf"), float("-inf")): + invalid_decision = StrategyDecision( + positions=(PositionTarget(symbol=symbols[0], target_weight=invalid_weight),), + budgets=(BudgetIntent(name="portfolio", amount=1.0),), + ) + with patch( + "crypto_strategies.entrypoints._common._qpk_assess_with_evidence", + return_value=RiskGateResult( + decision=invalid_decision, + assessment=permissive_assessment, + ), + ): + invalid_result = apply_risk_gate(invalid_decision, ctx=ctx) + + assert invalid_result.positions == () + assert invalid_result.budgets == () + assert "rejected:overexposed" in invalid_result.risk_flags diff --git a/tests/test_entrypoints.py b/tests/test_entrypoints.py index 5dfcd38..532a9c5 100644 --- a/tests/test_entrypoints.py +++ b/tests/test_entrypoints.py @@ -1,5 +1,6 @@ from __future__ import annotations +from datetime import datetime, timedelta, timezone from types import SimpleNamespace import unittest from unittest.mock import patch @@ -9,6 +10,32 @@ from crypto_strategies import get_strategy_entrypoint +def _synthetic_member_mandate(*symbols: str) -> dict[str, object]: + now = datetime.now(timezone.utc) + return { + "mandate_id": "synthetic_algorithm_equivalence_only", + "mandate_version": "test-v1", + "authority_receipt_sha256": "a" * 64, + "authority_scope": "RESEARCH_ONLY", + "strategy_profile": "synthetic_test_fixture", + "account_mode": "synthetic_test_fixture", + "effective_at": (now - timedelta(minutes=1)).isoformat().replace("+00:00", "Z"), + "expires_at": (now + timedelta(minutes=1)).isoformat().replace("+00:00", "Z"), + "max_snapshot_age_seconds": 300, + "effective_exposure_cap": 1.0, + "loss_budget": 1_000_000.0, + "product_caps": {symbol: 1.0 for symbol in symbols}, + "nominal_caps": {symbol: 1.0 for symbol in symbols}, + "product_leverage_factors": {symbol: 1 for symbol in symbols}, + "allowed_nonzero_assets": list(symbols), + "source_revision": "b371322b948e4298920a7d8613b155245dcd5f8d", + } + + +def _fresh_as_of() -> datetime: + return datetime.now(timezone.utc) + + class CryptoStrategyEntrypointTests(unittest.TestCase): def test_crypto_live_pool_rotation_entrypoint_resolves_pool_from_upstream_artifact(self) -> None: entrypoint = get_strategy_entrypoint("crypto_live_pool_rotation") @@ -184,6 +211,8 @@ def test_crypto_live_pool_rotation_entrypoint_uses_authoritative_upstream_pool(s state = { "trend_pool_version": "2026-03-15-core_major", "trend_pool_as_of_date": "2026-03-15", + "ETHUSDT": {"is_holding": True, "entry_price": 2500.0, "highest_price": 3000.0}, + "SOLUSDT": {"is_holding": True, "entry_price": 150.0, "highest_price": 180.0}, } upstream_pool = ["BNBUSDT", "ETHUSDT", "SOLUSDT"] expected_budgets = legacy_core.compute_allocation_budgets( @@ -229,7 +258,7 @@ def test_crypto_live_pool_rotation_entrypoint_uses_authoritative_upstream_pool(s "derived_indicators": trend_indicators, "benchmark_snapshot": btc_snapshot, "portfolio_snapshot": PortfolioSnapshot( - as_of="2026-04-06", + as_of=_fresh_as_of(), total_equity=account_metrics["total_equity"], buying_power=account_metrics["cash_usdt"], cash_balance=account_metrics["cash_usdt"], @@ -241,15 +270,26 @@ def test_crypto_live_pool_rotation_entrypoint_uses_authoritative_upstream_pool(s metadata={ "account_metrics": account_metrics, "cash_available_for_trading": account_metrics["cash_usdt"], + "observed_effective_exposure": 0.27, }, ), "universe_snapshot": upstream_pool, }, state=state, - artifacts={"trend_pool_contract": {"source": "explicit_artifact"}}, + artifacts={ + "trend_pool_contract": {"source": "explicit_artifact"}, + "mandate_provenance": _synthetic_member_mandate( + "BTCUSDT", "BNBUSDT", "ETHUSDT", "SOLUSDT" + ), + }, ) ) + self.assertEqual( + decision.diagnostics["member_risk_assessment"]["outcome"], + "APPROVE", + decision.diagnostics["member_risk_assessment"], + ) budget_map = {budget.name: budget.amount for budget in decision.budgets} self.assertAlmostEqual(budget_map["btc_core_dca_pool"], expected_budgets["dca_usdt_pool"]) self.assertAlmostEqual(budget_map["trend_rotation_pool"], expected_budgets["trend_usdt_pool"]) @@ -273,6 +313,11 @@ def test_crypto_live_pool_rotation_entrypoint_uses_authoritative_upstream_pool(s self.assertEqual(tuple(decision.diagnostics["eligible_buy_symbols"]), tuple(expected_eligible_buy_symbols)) self.assertEqual(decision.diagnostics["planned_trend_buys"], expected_planned_trend_buys) self.assertEqual(decision.diagnostics["sell_reasons"], {}) + self.assertEqual(decision.diagnostics["strategy_stop_evaluation"]["outcome"], "CLEAR") + self.assertEqual( + decision.diagnostics["strategy_stop_evaluation"]["decision_digest_sha256"], + decision.diagnostics["member_risk_assessment"]["decision_digest_sha256"], + ) self.assertAlmostEqual( decision.diagnostics["btc_base_order_usdt"], legacy_core.get_dynamic_btc_base_order(account_metrics["total_equity"]), @@ -323,7 +368,7 @@ def test_crypto_equity_combo_entrypoint_exposes_binance_execution_contract(self) }, "benchmark_snapshot": {"regime_on": True}, "portfolio_snapshot": PortfolioSnapshot( - as_of="2026-04-06", + as_of=_fresh_as_of(), total_equity=1000.0, buying_power=1000.0, cash_balance=1000.0, @@ -334,17 +379,28 @@ def test_crypto_equity_combo_entrypoint_exposes_binance_execution_contract(self) "trend_value": 0.0, "dca_value": 0.0, }, + "observed_effective_exposure": 0.0, }, ), "universe_snapshot": ("ETHUSDT", "SOLUSDT"), }, state={}, + artifacts={ + "mandate_provenance": _synthetic_member_mandate( + "BTCUSDT", "ETHUSDT", "SOLUSDT" + ) + }, ) ) - budget_map = {budget.name: budget.amount for budget in decision.budgets} - self.assertGreater(budget_map["trend_rotation_pool"], 0.0) - self.assertGreater(budget_map["btc_core_dca_pool"], 0.0) + self.assertEqual( + decision.diagnostics["member_risk_assessment"]["outcome"], + "APPROVE", + decision.diagnostics["member_risk_assessment"], + ) + self.assertEqual(decision.positions, ()) + self.assertEqual(decision.budgets, ()) + self.assertIn("rejected:strategy_concentration", decision.risk_flags) self.assertGreater(decision.diagnostics["btc_base_order_usdt"], 0.0) self.assertGreater(decision.diagnostics["btc_target_ratio"], 0.0) self.assertGreater(decision.diagnostics["trend_target_ratio"], 0.0) @@ -386,6 +442,183 @@ def test_crypto_live_pool_rotation_entrypoint_sets_regime_off_flag_when_btc_regi self.assertIn("regime_off", decision.risk_flags) self.assertIn("no_trend_candidates", decision.risk_flags) + self.assertEqual(decision.positions, ()) + self.assertEqual(decision.budgets, ()) + self.assertEqual(decision.diagnostics["member_risk_assessment"]["outcome"], "REJECT") + + def test_crypto_live_pool_rotation_missing_held_stop_input_is_no_order(self) -> None: + entrypoint = get_strategy_entrypoint("crypto_live_pool_rotation") + fake_core = SimpleNamespace( + compute_allocation_budgets=lambda *_args: { + "btc_target_ratio": 0.1, + "trend_target_ratio": 0.1, + "trend_usdt_pool": 10.0, + "dca_usdt_pool": 10.0, + }, + select_rotation_weights=lambda *_args, **_kwargs: { + "ETHUSDT": {"weight": 1.0, "relative_score": 1.0, "abs_momentum": 0.1} + }, + get_dynamic_btc_base_order=lambda _total_equity: 1.0, + allocate_trend_buy_budget=lambda *_args, **_kwargs: {}, + ) + fake_rotation = SimpleNamespace( + resolve_authoritative_rotation_pool=lambda *_args, **_kwargs: ["ETHUSDT"], + plan_trend_buys=lambda *_args, **_kwargs: ([], {}), + ) + def evaluate(prices, indicators): + now = _fresh_as_of() + return entrypoint.evaluate( + StrategyContext( + as_of=now, + market_data={ + "market_prices": prices, + "derived_indicators": {"ETHUSDT": indicators}, + "benchmark_snapshot": {"regime_on": True}, + "portfolio_snapshot": PortfolioSnapshot( + as_of=now, + total_equity=1000.0, + positions=( + Position(symbol="ETHUSDT", quantity=1.0, market_value=3000.0), + ), + metadata={ + "account_metrics": { + "total_equity": 1000.0, + "cash_usdt": 1000.0, + "trend_value": 0.0, + "dca_value": 0.0, + }, + "observed_effective_exposure": 0.0, + }, + ), + "universe_snapshot": (), + }, + state={ + "ETHUSDT": { + "is_holding": True, + "entry_price": 2800.0, + "highest_price": 3200.0, + } + }, + artifacts={ + "mandate_provenance": _synthetic_member_mandate( + "BTCUSDT", "ETHUSDT" + ) + }, + ) + ) + + with patch( + "crypto_strategies.entrypoints._load_legacy_modules", + return_value=(fake_core, fake_rotation), + ): + missing = evaluate({}, {"sma60": 2600.0}) + triggered = evaluate( + {"ETHUSDT": 2000.0}, + {"atr14": 100.0, "sma60": 2600.0}, + ) + + for decision in (missing, triggered): + with self.subTest(decision=decision): + self.assertEqual(decision.positions, ()) + self.assertEqual(decision.budgets, ()) + self.assertEqual( + decision.diagnostics["strategy_stop_evaluation"]["outcome"], + "TRIGGERED", + ) + self.assertEqual( + decision.diagnostics["strategy_stop_evaluation"]["action_result"], + "BLOCKED", + ) + self.assertIn("rejected:strategy_stop_input", missing.risk_flags) + + def test_crypto_live_pool_rotation_invalid_held_highest_price_is_no_order(self) -> None: + entrypoint = get_strategy_entrypoint("crypto_live_pool_rotation") + fake_core = SimpleNamespace( + compute_allocation_budgets=lambda *_args: { + "btc_target_ratio": 0.1, + "trend_target_ratio": 0.1, + "trend_usdt_pool": 10.0, + "dca_usdt_pool": 10.0, + }, + select_rotation_weights=lambda *_args, **_kwargs: { + "ETHUSDT": {"weight": 1.0, "relative_score": 1.0, "abs_momentum": 0.1} + }, + get_dynamic_btc_base_order=lambda _total_equity: 1.0, + allocate_trend_buy_budget=lambda *_args, **_kwargs: {}, + ) + fake_rotation = SimpleNamespace( + resolve_authoritative_rotation_pool=lambda *_args, **_kwargs: ["ETHUSDT"], + plan_trend_buys=lambda *_args, **_kwargs: ([], {}), + ) + missing = object() + + def evaluate(highest_price): + now = _fresh_as_of() + symbol_state = {"is_holding": True, "entry_price": 2800.0} + if highest_price is not missing: + symbol_state["highest_price"] = highest_price + return entrypoint.evaluate( + StrategyContext( + as_of=now, + market_data={ + "market_prices": {"ETHUSDT": 3000.0}, + "derived_indicators": { + "ETHUSDT": {"atr14": 100.0, "sma60": 2600.0} + }, + "benchmark_snapshot": {"regime_on": True}, + "portfolio_snapshot": PortfolioSnapshot( + as_of=now, + total_equity=1000.0, + positions=( + Position(symbol="ETHUSDT", quantity=1.0, market_value=3000.0), + ), + metadata={ + "account_metrics": { + "total_equity": 1000.0, + "cash_usdt": 1000.0, + "trend_value": 0.0, + "dca_value": 0.0, + }, + "observed_effective_exposure": 0.0, + }, + ), + "universe_snapshot": (), + }, + state={"ETHUSDT": symbol_state}, + artifacts={ + "mandate_provenance": _synthetic_member_mandate( + "BTCUSDT", "ETHUSDT" + ) + }, + ) + ) + + with patch( + "crypto_strategies.entrypoints._load_legacy_modules", + return_value=(fake_core, fake_rotation), + ): + for highest_price in ( + missing, + None, + float("nan"), + float("inf"), + 0.0, + -1.0, + 2700.0, + ): + with self.subTest(highest_price=highest_price): + decision = evaluate(highest_price) + self.assertEqual(decision.positions, ()) + self.assertEqual(decision.budgets, ()) + self.assertIn("rejected:strategy_stop_input", decision.risk_flags) + self.assertEqual( + decision.diagnostics["strategy_stop_evaluation"]["outcome"], + "TRIGGERED", + ) + self.assertEqual( + decision.diagnostics["strategy_stop_evaluation"]["action_result"], + "BLOCKED", + ) if __name__ == "__main__": diff --git a/tests/test_rotation_authority.py b/tests/test_rotation_authority.py index 99894df..01512ee 100644 --- a/tests/test_rotation_authority.py +++ b/tests/test_rotation_authority.py @@ -3,10 +3,64 @@ from datetime import datetime, timezone import unittest -from crypto_strategies.strategies.crypto_live_pool_rotation.rotation import resolve_authoritative_rotation_pool +from crypto_strategies.strategies.crypto_live_pool_rotation.rotation import ( + build_strategy_stop_evaluation, + evaluate_held_trend_stops, + resolve_authoritative_rotation_pool, +) class RotationAuthorityTests(unittest.TestCase): + def test_held_symbol_missing_price_or_atr_blocks_stop_clear(self) -> None: + state = { + "ETHUSDT": { + "is_holding": True, + "entry_price": 2800.0, + "highest_price": 3200.0, + } + } + + for prices, indicators in ( + ({}, {"ETHUSDT": {"atr14": 100.0, "sma60": 2600.0}}), + ({"ETHUSDT": 3000.0}, {"ETHUSDT": {"atr14": float("nan"), "sma60": 2600.0}}), + ): + with self.subTest(prices=prices, indicators=indicators): + reasons, input_blocked = evaluate_held_trend_stops( + state, + held_symbols=("ETHUSDT",), + prices=prices, + indicators_map=indicators, + selected_candidates={"ETHUSDT": {}}, + atr_multiplier=2.5, + get_symbol_trade_state_fn=lambda current_state, symbol: current_state[symbol], + set_symbol_trade_state_fn=lambda *_args: None, + translate_fn=lambda key, **_kwargs: key, + ) + + self.assertTrue(input_blocked) + self.assertEqual(reasons, {"ETHUSDT": "trend_sell_reason_missing_stop_input"}) + + def test_strategy_stop_evaluation_is_versioned_and_digest_bound(self) -> None: + evaluation = build_strategy_stop_evaluation( + evaluated_at="2026-08-04T08:00:00Z", + decision_digest_sha256="b" * 64, + outcome="TRIGGERED", + action_result="BLOCKED", + ) + + self.assertEqual( + evaluation, + { + "evaluated": True, + "policy_id": "crypto_live_pool_rotation.executable_stop", + "policy_version": "v1", + "evaluated_at": "2026-08-04T08:00:00Z", + "decision_digest_sha256": "b" * 64, + "outcome": "TRIGGERED", + "action_result": "BLOCKED", + }, + ) + def test_resolve_authoritative_rotation_pool_uses_ordered_upstream_symbols(self) -> None: state = { "trend_pool_version": "2026-03-15-core_major", diff --git a/uv.lock b/uv.lock index 6bcdb46..7dbf742 100644 --- a/uv.lock +++ b/uv.lock @@ -11,9 +11,9 @@ dependencies = [ ] [package.metadata] -requires-dist = [{ name = "quant-platform-kit", git = "https://github.com/QuantStrategyLab/QuantPlatformKit.git?rev=776fe71e57e2924fcd1c73126f41d244242240bb" }] +requires-dist = [{ name = "quant-platform-kit", git = "https://github.com/QuantStrategyLab/QuantPlatformKit.git?rev=b371322b948e4298920a7d8613b155245dcd5f8d" }] [[package]] name = "quant-platform-kit" version = "0.10.0" -source = { git = "https://github.com/QuantStrategyLab/QuantPlatformKit.git?rev=776fe71e57e2924fcd1c73126f41d244242240bb#776fe71e57e2924fcd1c73126f41d244242240bb" } +source = { git = "https://github.com/QuantStrategyLab/QuantPlatformKit.git?rev=b371322b948e4298920a7d8613b155245dcd5f8d#b371322b948e4298920a7d8613b155245dcd5f8d" } From 1fa75756d4d461a11ee92714522e431ddd8e6160 Mon Sep 17 00:00:00 2001 From: Pigbibi <20649888+Pigbibi@users.noreply.github.com> Date: Tue, 4 Aug 2026 18:18:58 +0800 Subject: [PATCH 2/3] fix(risk): close held-stop fail-closed gaps Co-Authored-By: Codex --- src/crypto_strategies/entrypoints/__init__.py | 109 ++++++----- .../crypto_live_pool_rotation/rotation.py | 13 +- tests/test_entrypoints.py | 177 ++++++++++++++++++ tests/test_rotation_authority.py | 64 +++++++ 4 files changed, 316 insertions(+), 47 deletions(-) diff --git a/src/crypto_strategies/entrypoints/__init__.py b/src/crypto_strategies/entrypoints/__init__.py index 68fa383..d43a108 100644 --- a/src/crypto_strategies/entrypoints/__init__.py +++ b/src/crypto_strategies/entrypoints/__init__.py @@ -121,7 +121,13 @@ def _set_symbol_trade_state(state, symbol, symbol_state): return _get_symbol_trade_state, _set_symbol_trade_state -def _resolve_held_risk_symbols(ctx: StrategyContext, state): +def _resolve_held_risk_symbols( + ctx: StrategyContext, + state, + *, + trend_universe_symbols, + get_symbol_trade_state_fn, +): held = { str(symbol).strip().upper() for symbol, payload in state.items() @@ -130,8 +136,15 @@ def _resolve_held_risk_symbols(ctx: StrategyContext, state): and str(symbol).strip().upper() != "BTCUSDT" } snapshot = _resolve_portfolio_snapshot(ctx) + candidate_symbols = { + str(symbol).strip().upper() + for symbol in trend_universe_symbols + if str(symbol).strip().upper() != "BTCUSDT" + } for position in getattr(snapshot, "positions", ()) or (): symbol = str(getattr(position, "symbol", "")).strip().upper() + if symbol and symbol != "BTCUSDT": + candidate_symbols.add(symbol) quantity = getattr(position, "quantity", 0.0) market_value = getattr(position, "market_value", 0.0) values = (quantity, market_value) @@ -143,6 +156,10 @@ def _resolve_held_risk_symbols(ctx: StrategyContext, state): for value in values ): held.add(symbol) + for symbol in candidate_symbols: + symbol_state = get_symbol_trade_state_fn(state, symbol) + if isinstance(symbol_state, Mapping) and symbol_state.get("is_holding"): + held.add(symbol) return tuple(sorted(held)) @@ -189,10 +206,12 @@ def evaluate_crypto_live_pool_rotation(ctx: StrategyContext) -> StrategyDecision weight_mode=str(config.get("weight_mode", "inverse_vol")), ) - atr_multiplier = float(config.get("atr_multiplier", 2.5)) + atr_multiplier = config.get("atr_multiplier", 2.5) held_risk_symbols = _resolve_held_risk_symbols( ctx, working_state, + trend_universe_symbols=trend_universe_symbols, + get_symbol_trade_state_fn=get_symbol_trade_state_fn, ) sell_reasons, stop_input_blocked = evaluate_held_trend_stops( working_state, @@ -206,50 +225,56 @@ def evaluate_crypto_live_pool_rotation(ctx: StrategyContext) -> StrategyDecision translate_fn=translator, ) - eligible_buy_symbols, planned_trend_buys = legacy_rotation.plan_trend_buys( - working_state, - runtime_trend_universe={symbol: {"base_asset": symbol[:-4]} for symbol in trend_universe_symbols}, - selected_candidates=selected_candidates, - trend_indicators=indicators_map, - prices=prices, - available_trend_buy_budget=float(budgets["trend_usdt_pool"]), - allow_new_trend_entries=bool(config.get("allow_new_trend_entries", True)), - get_symbol_trade_state_fn=get_symbol_trade_state_fn, - allocate_trend_buy_budget_fn=legacy_core.allocate_trend_buy_budget, - ) - - positions = [ - PositionTarget( - symbol="BTCUSDT", - target_weight=float(budgets["btc_target_ratio"]), - role="core", + if stop_input_blocked: + eligible_buy_symbols, planned_trend_buys = (), {} + positions = [] + budget_intents = () + else: + eligible_buy_symbols, planned_trend_buys = legacy_rotation.plan_trend_buys( + working_state, + runtime_trend_universe={ + symbol: {"base_asset": symbol[:-4]} for symbol in trend_universe_symbols + }, + selected_candidates=selected_candidates, + trend_indicators=indicators_map, + prices=prices, + available_trend_buy_budget=float(budgets["trend_usdt_pool"]), + allow_new_trend_entries=bool(config.get("allow_new_trend_entries", True)), + get_symbol_trade_state_fn=get_symbol_trade_state_fn, + allocate_trend_buy_budget_fn=legacy_core.allocate_trend_buy_budget, ) - ] - trend_target_ratio = float(budgets["trend_target_ratio"]) - for symbol, payload in sorted(selected_candidates.items()): - if symbol in sell_reasons: - continue - positions.append( + positions = [ PositionTarget( - symbol=symbol, - target_weight=trend_target_ratio * float(payload["weight"]), - role="trend_rotation", + symbol="BTCUSDT", + target_weight=float(budgets["btc_target_ratio"]), + role="core", + ) + ] + trend_target_ratio = float(budgets["trend_target_ratio"]) + for symbol, payload in sorted(selected_candidates.items()): + if symbol in sell_reasons: + continue + positions.append( + PositionTarget( + symbol=symbol, + target_weight=trend_target_ratio * float(payload["weight"]), + role="trend_rotation", + ) ) - ) - budget_intents = ( - BudgetIntent( - name="btc_core_dca_pool", - symbol="BTCUSDT", - amount=float(budgets["dca_usdt_pool"]), - purpose="btc_core_accumulation", - ), - BudgetIntent( - name="trend_rotation_pool", - amount=float(budgets["trend_usdt_pool"]), - purpose="trend_rotation", - ), - ) + budget_intents = ( + BudgetIntent( + name="btc_core_dca_pool", + symbol="BTCUSDT", + amount=float(budgets["dca_usdt_pool"]), + purpose="btc_core_accumulation", + ), + BudgetIntent( + name="trend_rotation_pool", + amount=float(budgets["trend_usdt_pool"]), + purpose="trend_rotation", + ), + ) risk_flags: tuple[str, ...] = () if not btc_snapshot.get("regime_on"): diff --git a/src/crypto_strategies/strategies/crypto_live_pool_rotation/rotation.py b/src/crypto_strategies/strategies/crypto_live_pool_rotation/rotation.py index 447a2d8..a344076 100644 --- a/src/crypto_strategies/strategies/crypto_live_pool_rotation/rotation.py +++ b/src/crypto_strategies/strategies/crypto_live_pool_rotation/rotation.py @@ -33,22 +33,25 @@ def evaluate_held_trend_stops( """Evaluate every held risk symbol; incomplete inputs block CLEAR.""" sell_reasons = {} input_blocked = False + valid_atr_multiplier = _finite_number(atr_multiplier) for symbol in _normalize_symbol_list(held_symbols): symbol_state = get_symbol_trade_state_fn(state, symbol) - persisted_symbol_state = state.get(symbol) if isinstance(state, Mapping) else None + if not isinstance(symbol_state, Mapping): + symbol_state = {} indicators = indicators_map.get(symbol) curr_price = _finite_number(prices.get(symbol)) atr = _finite_number(indicators.get("atr14")) if isinstance(indicators, Mapping) else None sma60 = _finite_number(indicators.get("sma60")) if isinstance(indicators, Mapping) else None entry_price = _finite_number(symbol_state.get("entry_price")) highest_price = ( - _finite_number(persisted_symbol_state.get("highest_price")) - if isinstance(persisted_symbol_state, Mapping) - and "highest_price" in persisted_symbol_state + _finite_number(symbol_state.get("highest_price")) + if "highest_price" in symbol_state else None ) if ( not symbol_state.get("is_holding") + or valid_atr_multiplier is None + or valid_atr_multiplier <= 0.0 or curr_price is None or atr is None or sma60 is None @@ -67,7 +70,7 @@ def evaluate_held_trend_stops( curr_price, indicators, selected_candidates, - atr_multiplier, + valid_atr_multiplier, get_symbol_trade_state_fn=get_symbol_trade_state_fn, set_symbol_trade_state_fn=set_symbol_trade_state_fn, translate_fn=translate_fn, diff --git a/tests/test_entrypoints.py b/tests/test_entrypoints.py index 532a9c5..a39bdb5 100644 --- a/tests/test_entrypoints.py +++ b/tests/test_entrypoints.py @@ -531,6 +531,183 @@ def evaluate(prices, indicators): ) self.assertIn("rejected:strategy_stop_input", missing.risk_flags) + def test_crypto_live_pool_rotation_blocked_stop_skips_buy_planning(self) -> None: + entrypoint = get_strategy_entrypoint("crypto_live_pool_rotation") + buy_plan_calls: list[object] = [] + + def plan_trend_buys(*args, **kwargs): + buy_plan_calls.append((args, kwargs)) + raise AssertionError("buy planning must not run after blocked stop input") + + fake_core = SimpleNamespace( + compute_allocation_budgets=lambda *_args: { + "btc_target_ratio": 0.1, + "trend_target_ratio": 0.1, + "trend_usdt_pool": 10.0, + "dca_usdt_pool": 10.0, + }, + select_rotation_weights=lambda *_args, **_kwargs: { + "ETHUSDT": {"weight": 1.0, "relative_score": 1.0, "abs_momentum": 0.1} + }, + get_dynamic_btc_base_order=lambda _total_equity: 1.0, + allocate_trend_buy_budget=lambda *_args, **_kwargs: {}, + ) + fake_rotation = SimpleNamespace( + resolve_authoritative_rotation_pool=lambda *_args, **_kwargs: ["ETHUSDT"], + plan_trend_buys=plan_trend_buys, + ) + now = _fresh_as_of() + + with patch( + "crypto_strategies.entrypoints._load_legacy_modules", + return_value=(fake_core, fake_rotation), + ): + decision = entrypoint.evaluate( + StrategyContext( + as_of=now, + market_data={ + "market_prices": {}, + "derived_indicators": { + "ETHUSDT": {"atr14": 100.0, "sma60": 2600.0} + }, + "benchmark_snapshot": {"regime_on": True}, + "portfolio_snapshot": PortfolioSnapshot( + as_of=now, + total_equity=1000.0, + positions=( + Position(symbol="ETHUSDT", quantity=1.0, market_value=3000.0), + ), + metadata={ + "account_metrics": { + "total_equity": 1000.0, + "cash_usdt": 1000.0, + "trend_value": 0.0, + "dca_value": 0.0, + }, + "observed_effective_exposure": 0.0, + }, + ), + "universe_snapshot": ("ETHUSDT",), + }, + state={}, + artifacts={ + "mandate_provenance": _synthetic_member_mandate( + "BTCUSDT", "ETHUSDT" + ) + }, + ) + ) + + self.assertEqual(buy_plan_calls, []) + self.assertEqual(decision.positions, ()) + self.assertEqual(decision.budgets, ()) + self.assertIn("rejected:strategy_stop_input", decision.risk_flags) + self.assertEqual(decision.diagnostics["eligible_buy_symbols"], ()) + self.assertEqual(decision.diagnostics["planned_trend_buys"], {}) + self.assertEqual( + decision.diagnostics["strategy_stop_evaluation"]["outcome"], + "TRIGGERED", + ) + self.assertEqual( + decision.diagnostics["strategy_stop_evaluation"]["action_result"], + "BLOCKED", + ) + + def test_crypto_live_pool_rotation_discovers_nested_custom_held_state(self) -> None: + entrypoint = get_strategy_entrypoint("crypto_live_pool_rotation") + state_get_calls: list[str] = [] + + def get_symbol_trade_state(state, symbol): + state_get_calls.append(symbol) + symbol_state = state.get("trade_states", {}).get(symbol) + if not isinstance(symbol_state, dict): + return {"is_holding": False, "entry_price": 0.0, "highest_price": 0.0} + return dict(symbol_state) + + def set_symbol_trade_state(state, symbol, symbol_state): + state.setdefault("trade_states", {})[symbol] = dict(symbol_state) + + fake_core = SimpleNamespace( + compute_allocation_budgets=lambda *_args: { + "btc_target_ratio": 0.1, + "trend_target_ratio": 0.1, + "trend_usdt_pool": 10.0, + "dca_usdt_pool": 10.0, + }, + select_rotation_weights=lambda *_args, **_kwargs: { + "ETHUSDT": {"weight": 1.0, "relative_score": 1.0, "abs_momentum": 0.1} + }, + get_dynamic_btc_base_order=lambda _total_equity: 1.0, + allocate_trend_buy_budget=lambda *_args, **_kwargs: {}, + ) + fake_rotation = SimpleNamespace( + resolve_authoritative_rotation_pool=lambda *_args, **_kwargs: ["ETHUSDT"], + plan_trend_buys=lambda *_args, **_kwargs: ([], {}), + ) + now = _fresh_as_of() + + with patch( + "crypto_strategies.entrypoints._load_legacy_modules", + return_value=(fake_core, fake_rotation), + ): + decision = entrypoint.evaluate( + StrategyContext( + as_of=now, + market_data={ + "market_prices": {"ETHUSDT": 3000.0}, + "derived_indicators": { + "ETHUSDT": {"atr14": 100.0, "sma60": 2600.0} + }, + "benchmark_snapshot": {"regime_on": True}, + "portfolio_snapshot": PortfolioSnapshot( + as_of=now, + total_equity=1000.0, + metadata={ + "account_metrics": { + "total_equity": 1000.0, + "cash_usdt": 1000.0, + "trend_value": 0.0, + "dca_value": 0.0, + }, + "observed_effective_exposure": 0.0, + }, + ), + "universe_snapshot": ("ETHUSDT",), + }, + state={ + "trade_states": { + "ETHUSDT": { + "is_holding": True, + "entry_price": 2800.0, + } + } + }, + runtime_config={ + "get_symbol_trade_state_fn": get_symbol_trade_state, + "set_symbol_trade_state_fn": set_symbol_trade_state, + }, + artifacts={ + "mandate_provenance": _synthetic_member_mandate( + "BTCUSDT", "ETHUSDT" + ) + }, + ) + ) + + self.assertGreaterEqual(state_get_calls.count("ETHUSDT"), 2) + self.assertEqual(decision.positions, ()) + self.assertEqual(decision.budgets, ()) + self.assertIn("rejected:strategy_stop_input", decision.risk_flags) + self.assertIn("ETHUSDT", decision.diagnostics["sell_reasons"]) + self.assertEqual( + decision.diagnostics["strategy_stop_evaluation"]["outcome"], + "TRIGGERED", + ) + self.assertEqual( + decision.diagnostics["strategy_stop_evaluation"]["action_result"], + "BLOCKED", + ) + def test_crypto_live_pool_rotation_invalid_held_highest_price_is_no_order(self) -> None: entrypoint = get_strategy_entrypoint("crypto_live_pool_rotation") fake_core = SimpleNamespace( diff --git a/tests/test_rotation_authority.py b/tests/test_rotation_authority.py index 01512ee..e5aad5b 100644 --- a/tests/test_rotation_authority.py +++ b/tests/test_rotation_authority.py @@ -40,6 +40,70 @@ def test_held_symbol_missing_price_or_atr_blocks_stop_clear(self) -> None: self.assertTrue(input_blocked) self.assertEqual(reasons, {"ETHUSDT": "trend_sell_reason_missing_stop_input"}) + def test_invalid_atr_multiplier_blocks_stop_clear(self) -> None: + state = { + "ETHUSDT": { + "is_holding": True, + "entry_price": 2800.0, + "highest_price": 3200.0, + } + } + + for atr_multiplier in ( + True, + False, + None, + "2.5", + float("nan"), + float("inf"), + float("-inf"), + 0.0, + -1.0, + ): + with self.subTest(atr_multiplier=atr_multiplier): + reasons, input_blocked = evaluate_held_trend_stops( + state, + held_symbols=("ETHUSDT",), + prices={"ETHUSDT": 3150.0}, + indicators_map={"ETHUSDT": {"atr14": 100.0, "sma60": 2600.0}}, + selected_candidates={"ETHUSDT": {}}, + atr_multiplier=atr_multiplier, + get_symbol_trade_state_fn=lambda current_state, symbol: current_state[symbol], + set_symbol_trade_state_fn=lambda *_args: None, + translate_fn=lambda key, **_kwargs: key, + ) + + self.assertTrue(input_blocked) + self.assertEqual(reasons, {"ETHUSDT": "trend_sell_reason_missing_stop_input"}) + + def test_held_stop_reads_persisted_state_through_custom_helper(self) -> None: + state = { + "trade_states": { + "ETHUSDT": { + "is_holding": True, + "entry_price": 2800.0, + "highest_price": 3200.0, + } + } + } + + reasons, input_blocked = evaluate_held_trend_stops( + state, + held_symbols=("ETHUSDT",), + prices={"ETHUSDT": 3150.0}, + indicators_map={"ETHUSDT": {"atr14": 100.0, "sma60": 2600.0}}, + selected_candidates={"ETHUSDT": {}}, + atr_multiplier=2.5, + get_symbol_trade_state_fn=lambda current_state, symbol: current_state[ + "trade_states" + ][symbol], + set_symbol_trade_state_fn=lambda *_args: None, + translate_fn=lambda key, **_kwargs: key, + ) + + self.assertFalse(input_blocked) + self.assertEqual(reasons, {}) + def test_strategy_stop_evaluation_is_versioned_and_digest_bound(self) -> None: evaluation = build_strategy_stop_evaluation( evaluated_at="2026-08-04T08:00:00Z", From 07eaa322ccf231737e0c671cb7a6ae348db2a5e7 Mon Sep 17 00:00:00 2001 From: Pigbibi <20649888+Pigbibi@users.noreply.github.com> Date: Tue, 4 Aug 2026 18:32:41 +0800 Subject: [PATCH 3/3] fix(risk): reject nonpositive held ATR Co-Authored-By: Codex --- .../crypto_live_pool_rotation/rotation.py | 1 + tests/test_entrypoints.py | 22 ++++++++++++++++--- 2 files changed, 20 insertions(+), 3 deletions(-) diff --git a/src/crypto_strategies/strategies/crypto_live_pool_rotation/rotation.py b/src/crypto_strategies/strategies/crypto_live_pool_rotation/rotation.py index a344076..1ce1854 100644 --- a/src/crypto_strategies/strategies/crypto_live_pool_rotation/rotation.py +++ b/src/crypto_strategies/strategies/crypto_live_pool_rotation/rotation.py @@ -54,6 +54,7 @@ def evaluate_held_trend_stops( or valid_atr_multiplier <= 0.0 or curr_price is None or atr is None + or atr <= 0.0 or sma60 is None or entry_price is None or entry_price <= 0.0 diff --git a/tests/test_entrypoints.py b/tests/test_entrypoints.py index a39bdb5..23fe5ed 100644 --- a/tests/test_entrypoints.py +++ b/tests/test_entrypoints.py @@ -448,6 +448,12 @@ def test_crypto_live_pool_rotation_entrypoint_sets_regime_off_flag_when_btc_regi def test_crypto_live_pool_rotation_missing_held_stop_input_is_no_order(self) -> None: entrypoint = get_strategy_entrypoint("crypto_live_pool_rotation") + buy_plan_calls: list[object] = [] + + def plan_trend_buys(*args, **kwargs): + buy_plan_calls.append((args, kwargs)) + return [], {} + fake_core = SimpleNamespace( compute_allocation_budgets=lambda *_args: { "btc_target_ratio": 0.1, @@ -463,7 +469,7 @@ def test_crypto_live_pool_rotation_missing_held_stop_input_is_no_order(self) -> ) fake_rotation = SimpleNamespace( resolve_authoritative_rotation_pool=lambda *_args, **_kwargs: ["ETHUSDT"], - plan_trend_buys=lambda *_args, **_kwargs: ([], {}), + plan_trend_buys=plan_trend_buys, ) def evaluate(prices, indicators): now = _fresh_as_of() @@ -516,8 +522,16 @@ def evaluate(prices, indicators): {"ETHUSDT": 2000.0}, {"atr14": 100.0, "sma60": 2600.0}, ) + buy_plan_calls_after_valid_stop = len(buy_plan_calls) + invalid_atr = tuple( + evaluate( + {"ETHUSDT": 3200.0}, + {"atr14": atr14, "sma60": 2600.0}, + ) + for atr14 in (0.0, -1.0) + ) - for decision in (missing, triggered): + for decision in (missing, triggered, *invalid_atr): with self.subTest(decision=decision): self.assertEqual(decision.positions, ()) self.assertEqual(decision.budgets, ()) @@ -529,7 +543,9 @@ def evaluate(prices, indicators): decision.diagnostics["strategy_stop_evaluation"]["action_result"], "BLOCKED", ) - self.assertIn("rejected:strategy_stop_input", missing.risk_flags) + for decision in (missing, *invalid_atr): + self.assertIn("rejected:strategy_stop_input", decision.risk_flags) + self.assertEqual(len(buy_plan_calls), buy_plan_calls_after_valid_stop) def test_crypto_live_pool_rotation_blocked_stop_skips_buy_planning(self) -> None: entrypoint = get_strategy_entrypoint("crypto_live_pool_rotation")