From b5c9ae4c613de2b05440cd4bcb3ce8288c61af8d Mon Sep 17 00:00:00 2001 From: QuantStrategyLab QPK Sync Date: Tue, 4 Aug 2026 07:30:30 +0000 Subject: [PATCH 1/4] chore(deps): align QPK pin to b371322b948e Automated downstream QPK pin update after QPK_PIN landed on main. Co-Authored-By: Claude --- pyproject.toml | 2 +- qsl.toml | 2 +- uv.lock | 4 ++-- 3 files changed, 4 insertions(+), 4 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/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 3f32e3332e89d40d7c19332fba219bc560d6fa43 Mon Sep 17 00:00:00 2001 From: Pigbibi <20649888+Pigbibi@users.noreply.github.com> Date: Tue, 4 Aug 2026 15:51:15 +0800 Subject: [PATCH 2/4] feat(risk): bind member and stop evidence Co-Authored-By: Codex --- src/crypto_strategies/entrypoints/__init__.py | 98 +++++++++--- src/crypto_strategies/entrypoints/_common.py | 44 ++++-- .../crypto_live_pool_rotation/rotation.py | 72 +++++++++ tests/test_entrypoint_risk_gate.py | 62 +++++++- tests/test_entrypoints.py | 148 +++++++++++++++++- tests/test_rotation_authority.py | 56 ++++++- 6 files changed, 435 insertions(+), 45 deletions(-) diff --git a/src/crypto_strategies/entrypoints/__init__.py b/src/crypto_strategies/entrypoints/__init__.py index 9376d9f..ce334ae 100644 --- a/src/crypto_strategies/entrypoints/__init__.py +++ b/src/crypto_strategies/entrypoints/__init__.py @@ -19,6 +19,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 +120,23 @@ 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, universe_symbols, get_symbol_trade_state_fn): + available = {str(symbol).strip().upper() for symbol in universe_symbols} + held = { + symbol + for symbol in available + if get_symbol_trade_state_fn(state, symbol).get("is_holding") + } + 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) + if symbol in available and (quantity or market_value): + 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 +180,24 @@ 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, + trend_universe_symbols, + get_symbol_trade_state_fn, + ) + 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 +220,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 +279,41 @@ 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 stop_input_blocked: + 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 = ( + "COMPLETED" + if member_assessment["outcome"] == "APPROVE" and not stop_input_blocked + else "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, @@ -350,7 +402,7 @@ def evaluate_crypto_btc_dca(ctx: StrategyContext) -> StrategyDecision: risk_flags=risk_flags, diagnostics=diagnostics, ) - decision = apply_risk_gate(decision, ctx=ctx, max_single_weight=0.50) + decision = apply_risk_gate(decision, ctx=ctx) record_strategy_decision( ctx, decision, @@ -440,7 +492,7 @@ def evaluate_crypto_trend_rotation(ctx: StrategyContext) -> StrategyDecision: risk_flags=risk_flags, diagnostics=diagnostics, ) - decision = apply_risk_gate(decision, ctx=ctx, max_single_weight=0.30) + decision = apply_risk_gate(decision, ctx=ctx) record_strategy_decision( ctx, decision, @@ -636,7 +688,7 @@ def evaluate_crypto_equity_combo(ctx: StrategyContext) -> StrategyDecision: risk_flags=risk_flags, diagnostics=diagnostics, ) - decision = apply_risk_gate(decision, ctx=ctx, max_single_weight=0.30) + decision = apply_risk_gate(decision, ctx=ctx) record_strategy_decision( ctx, decision, diff --git a/src/crypto_strategies/entrypoints/_common.py b/src/crypto_strategies/entrypoints/_common.py index c022612..578a413 100644 --- a/src/crypto_strategies/entrypoints/_common.py +++ b/src/crypto_strategies/entrypoints/_common.py @@ -2,12 +2,13 @@ import logging 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 +50,15 @@ 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, 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 +68,25 @@ 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 ())) + ) + return StrategyDecision( + positions=result.decision.positions, + budgets=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..d8d4799 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,79 @@ 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) + 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 + if not symbol_state.get("is_holding") or curr_price is None or atr is None or sma60 is None: + 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..41a5b6d 100644 --- a/tests/test_entrypoint_risk_gate.py +++ b/tests/test_entrypoint_risk_gate.py @@ -1,13 +1,34 @@ from __future__ import annotations -from datetime import datetime, timezone +from datetime import datetime, timedelta, timezone from quant_platform_kit.common.models import PortfolioSnapshot, Position -from quant_platform_kit.strategy_contracts import PositionTarget, StrategyContext, StrategyDecision +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 +42,39 @@ 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) diff --git a/tests/test_entrypoints.py b/tests/test_entrypoints.py index 5dfcd38..189a1a7 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,14 +379,25 @@ 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" + ) + }, ) ) + 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.assertGreater(budget_map["trend_rotation_pool"], 0.0) self.assertGreater(budget_map["btc_core_dca_pool"], 0.0) @@ -386,6 +442,92 @@ 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: ([], {}), + ) + 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": {"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={ + "ETHUSDT": { + "is_holding": True, + "entry_price": 2800.0, + "highest_price": 3200.0, + } + }, + artifacts={ + "mandate_provenance": _synthetic_member_mandate( + "BTCUSDT", "ETHUSDT" + ) + }, + ) + ) + + self.assertEqual(decision.positions, ()) + self.assertEqual(decision.budgets, ()) + self.assertIn("rejected:strategy_stop_input", decision.risk_flags) + self.assertEqual( + decision.diagnostics["strategy_stop_evaluation"], + { + "evaluated": True, + "policy_id": "crypto_live_pool_rotation.executable_stop", + "policy_version": "v1", + "evaluated_at": decision.diagnostics["member_risk_assessment"]["evaluated_at"], + "decision_digest_sha256": decision.diagnostics["member_risk_assessment"][ + "decision_digest_sha256" + ], + "outcome": "TRIGGERED", + "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", From d82c25d2efc212c979783cef6f588a854815792e Mon Sep 17 00:00:00 2001 From: Pigbibi <20649888+Pigbibi@users.noreply.github.com> Date: Tue, 4 Aug 2026 16:15:39 +0800 Subject: [PATCH 3/4] fix(risk): close stop evidence findings Co-Authored-By: Codex --- src/crypto_strategies/entrypoints/__init__.py | 37 ++++++----- src/crypto_strategies/entrypoints/_common.py | 21 ++++++- tests/test_entrypoint_risk_gate.py | 37 +++++++++++ tests/test_entrypoints.py | 62 ++++++++++--------- 4 files changed, 108 insertions(+), 49 deletions(-) diff --git a/src/crypto_strategies/entrypoints/__init__.py b/src/crypto_strategies/entrypoints/__init__.py index ce334ae..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, @@ -120,19 +121,27 @@ 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, universe_symbols, get_symbol_trade_state_fn): - available = {str(symbol).strip().upper() for symbol in universe_symbols} +def _resolve_held_risk_symbols(ctx: StrategyContext, state): held = { - symbol - for symbol in available - if get_symbol_trade_state_fn(state, symbol).get("is_holding") + 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) - if symbol in available and (quantity or market_value): + 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)) @@ -184,8 +193,6 @@ def evaluate_crypto_live_pool_rotation(ctx: StrategyContext) -> StrategyDecision held_risk_symbols = _resolve_held_risk_symbols( ctx, working_state, - trend_universe_symbols, - get_symbol_trade_state_fn, ) sell_reasons, stop_input_blocked = evaluate_held_trend_stops( working_state, @@ -283,7 +290,7 @@ def evaluate_crypto_live_pool_rotation(ctx: StrategyContext) -> StrategyDecision risk_flags=risk_flags + (("rejected:strategy_stop_input",) if stop_input_blocked else ()), diagnostics=diagnostics, ) - if stop_input_blocked: + if sell_reasons: decision = StrategyDecision( positions=(), budgets=(), @@ -295,11 +302,7 @@ def evaluate_crypto_live_pool_rotation(ctx: StrategyContext) -> StrategyDecision stop_outcome = "TRIGGERED" if sell_reasons else "CLEAR" stop_action_result = "NOT_REQUIRED" if stop_outcome == "TRIGGERED": - stop_action_result = ( - "COMPLETED" - if member_assessment["outcome"] == "APPROVE" and not stop_input_blocked - else "BLOCKED" - ) + stop_action_result = "BLOCKED" decision = StrategyDecision( positions=decision.positions, budgets=decision.budgets, @@ -402,7 +405,7 @@ def evaluate_crypto_btc_dca(ctx: StrategyContext) -> StrategyDecision: risk_flags=risk_flags, diagnostics=diagnostics, ) - decision = apply_risk_gate(decision, ctx=ctx) + decision = apply_risk_gate(decision, ctx=ctx, max_single_weight=0.50) record_strategy_decision( ctx, decision, @@ -492,7 +495,7 @@ def evaluate_crypto_trend_rotation(ctx: StrategyContext) -> StrategyDecision: risk_flags=risk_flags, diagnostics=diagnostics, ) - decision = apply_risk_gate(decision, ctx=ctx) + decision = apply_risk_gate(decision, ctx=ctx, max_single_weight=0.30) record_strategy_decision( ctx, decision, @@ -688,7 +691,7 @@ def evaluate_crypto_equity_combo(ctx: StrategyContext) -> StrategyDecision: risk_flags=risk_flags, diagnostics=diagnostics, ) - decision = apply_risk_gate(decision, ctx=ctx) + decision = apply_risk_gate(decision, ctx=ctx, max_single_weight=0.30) record_strategy_decision( ctx, decision, diff --git a/src/crypto_strategies/entrypoints/_common.py b/src/crypto_strategies/entrypoints/_common.py index 578a413..6a7ec9d 100644 --- a/src/crypto_strategies/entrypoints/_common.py +++ b/src/crypto_strategies/entrypoints/_common.py @@ -1,6 +1,7 @@ from __future__ import annotations import logging +import math from collections.abc import Mapping from dataclasses import asdict from typing import Any @@ -50,6 +51,7 @@ def apply_risk_gate( decision: StrategyDecision, *, ctx: StrategyContext | None = None, + max_single_weight: float | None = None, portfolio_snapshot: Any | None = None, market_data: Mapping[str, Any] | None = None, ) -> StrategyDecision: @@ -81,9 +83,24 @@ def apply_risk_gate( risk_flags = tuple( dict.fromkeys(tuple(decision.risk_flags or ()) + tuple(result.decision.risk_flags or ())) ) + 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 = any( + position.target_weight is not None + and ( + not math.isfinite(float(position.target_weight)) + or abs(float(position.target_weight)) > cap + ) + for position in result.decision.positions + ) + if strategy_concentration_rejected: + risk_flags = tuple(dict.fromkeys(risk_flags + ("rejected:strategy_concentration",))) return StrategyDecision( - positions=result.decision.positions, - budgets=result.decision.budgets, + positions=() if strategy_concentration_rejected else result.decision.positions, + budgets=() if strategy_concentration_rejected else result.decision.budgets, risk_flags=risk_flags, diagnostics={ **dict(result.decision.diagnostics or {}), diff --git a/tests/test_entrypoint_risk_gate.py b/tests/test_entrypoint_risk_gate.py index 41a5b6d..b77dc60 100644 --- a/tests/test_entrypoint_risk_gate.py +++ b/tests/test_entrypoint_risk_gate.py @@ -78,3 +78,40 @@ def test_apply_risk_gate_uses_member_evidence_and_zero_cap_clears_authority() -> 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 diff --git a/tests/test_entrypoints.py b/tests/test_entrypoints.py index 189a1a7..318a047 100644 --- a/tests/test_entrypoints.py +++ b/tests/test_entrypoints.py @@ -398,9 +398,9 @@ def test_crypto_equity_combo_entrypoint_exposes_binance_execution_contract(self) "APPROVE", decision.diagnostics["member_risk_assessment"], ) - 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.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) @@ -465,18 +465,14 @@ def test_crypto_live_pool_rotation_missing_held_stop_input_is_no_order(self) -> 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( + def evaluate(prices, indicators): + now = _fresh_as_of() + return entrypoint.evaluate( StrategyContext( as_of=now, market_data={ - "market_prices": {}, - "derived_indicators": {"ETHUSDT": {"sma60": 2600.0}}, + "market_prices": prices, + "derived_indicators": {"ETHUSDT": indicators}, "benchmark_snapshot": {"regime_on": True}, "portfolio_snapshot": PortfolioSnapshot( as_of=now, @@ -494,7 +490,7 @@ def test_crypto_live_pool_rotation_missing_held_stop_input_is_no_order(self) -> "observed_effective_exposure": 0.0, }, ), - "universe_snapshot": ("ETHUSDT",), + "universe_snapshot": (), }, state={ "ETHUSDT": { @@ -511,23 +507,29 @@ def test_crypto_live_pool_rotation_missing_held_stop_input_is_no_order(self) -> ) ) - self.assertEqual(decision.positions, ()) - self.assertEqual(decision.budgets, ()) - self.assertIn("rejected:strategy_stop_input", decision.risk_flags) - self.assertEqual( - decision.diagnostics["strategy_stop_evaluation"], - { - "evaluated": True, - "policy_id": "crypto_live_pool_rotation.executable_stop", - "policy_version": "v1", - "evaluated_at": decision.diagnostics["member_risk_assessment"]["evaluated_at"], - "decision_digest_sha256": decision.diagnostics["member_risk_assessment"][ - "decision_digest_sha256" - ], - "outcome": "TRIGGERED", - "action_result": "BLOCKED", - }, - ) + 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) if __name__ == "__main__": From d7a411dff4e3e9fc387cb4bf5ddf0832b1fd7f30 Mon Sep 17 00:00:00 2001 From: Pigbibi <20649888+Pigbibi@users.noreply.github.com> Date: Tue, 4 Aug 2026 17:26:41 +0800 Subject: [PATCH 4/4] fix: restore hard position count gate Co-Authored-By: Codex --- src/crypto_strategies/entrypoints/_common.py | 8 +++- tests/test_entrypoint_risk_gate.py | 40 ++++++++++++++++++++ 2 files changed, 46 insertions(+), 2 deletions(-) diff --git a/src/crypto_strategies/entrypoints/_common.py b/src/crypto_strategies/entrypoints/_common.py index 6a7ec9d..c46e480 100644 --- a/src/crypto_strategies/entrypoints/_common.py +++ b/src/crypto_strategies/entrypoints/_common.py @@ -98,9 +98,13 @@ def apply_risk_gate( ) 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",))) + strategy_rejected = strategy_concentration_rejected or strategy_position_count_rejected return StrategyDecision( - positions=() if strategy_concentration_rejected else result.decision.positions, - budgets=() if strategy_concentration_rejected else result.decision.budgets, + 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 {}), diff --git a/tests/test_entrypoint_risk_gate.py b/tests/test_entrypoint_risk_gate.py index b77dc60..2c0ccfe 100644 --- a/tests/test_entrypoint_risk_gate.py +++ b/tests/test_entrypoint_risk_gate.py @@ -115,3 +115,43 @@ def test_apply_risk_gate_preserves_stricter_strategy_concentration_cap() -> None 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