From 836d9ec8a5fbb68287dcc9c13a44c84dc2689b62 Mon Sep 17 00:00:00 2001 From: Pigbibi <20649888+Pigbibi@users.noreply.github.com> Date: Sun, 11 Oct 2026 20:37:04 +0900 Subject: [PATCH] report: observe portfolio stoploss coverage without changing decisions --- application/cycle_service.py | 14 +- docs/portfolio-stoploss-observation.md | 62 +++++++ main.py | 4 + portfolio_stoploss_observation.py | 142 +++++++++++++++ strategy_runtime.py | 10 ++ tests/test_cycle_service.py | 43 ++++- tests/test_portfolio_stoploss_observation.py | 171 +++++++++++++++++++ tests/test_strategy_runtime.py | 32 ++++ 8 files changed, 474 insertions(+), 4 deletions(-) create mode 100644 docs/portfolio-stoploss-observation.md create mode 100644 portfolio_stoploss_observation.py create mode 100644 tests/test_portfolio_stoploss_observation.py diff --git a/application/cycle_service.py b/application/cycle_service.py index 41e40de3..48b0136a 100644 --- a/application/cycle_service.py +++ b/application/cycle_service.py @@ -19,6 +19,7 @@ from application.execution_receipt_adapter import attach_execution_receipt_from_report from application.execution_kernel_adapter import consult_t2_unknown_new_cycle from application.portfolio_service import EARN_FORWARD_REASON_CODES +from portfolio_stoploss_observation import public_stoploss_observation, unevaluated_stoploss_observation from runtime_logging import RuntimeLogContext, emit_runtime_log from runtime_support import ( append_report_error, finalize_notification_delivery, acquire_runtime_state_owner, @@ -47,7 +48,10 @@ def try_record_platform_execution( _performance_monitor_logger.warning("platform_execution_record_failed") -def _record_risk_diagnostics(report, allocation): +def _record_risk_diagnostics(report, allocation, *, phase): + report.setdefault("diagnostics", {}).setdefault("portfolio_stoploss", {})[phase] = ( + public_stoploss_observation(allocation.get("portfolio_stoploss_observation")) + ) assessment = allocation.get("risk_assessment") if not isinstance(assessment, Mapping): return @@ -199,6 +203,10 @@ def execute_strategy_cycle( allow_new_trend_entries_on_degraded = cycle_settings.allow_new_trend_entries_on_degraded report = build_execution_report(runtime) + report.setdefault("diagnostics", {})["portfolio_stoploss"] = { + phase: unevaluated_stoploss_observation() + for phase in ("pre_trade", "post_trade") + } log_buffer = [] if not getattr(runtime, "standard_execution_permitted", True): report["execution_blocked_reason"] = "runtime_target_disabled" @@ -299,7 +307,7 @@ def execute_strategy_cycle( report["total_equity_usdt"] = total_equity report["trend_equity_usdt"] = trend_val_equity - _record_risk_diagnostics(report, allocation) + _record_risk_diagnostics(report, allocation, phase="pre_trade") if not allocation.get("execution_permitted", False): report["execution_blocked_reason"] = "risk_execution_not_permitted" @@ -413,7 +421,7 @@ def execute_strategy_cycle( report["total_equity_usdt"] = total_equity report["trend_equity_usdt"] = trend_val_equity - _record_risk_diagnostics(report, post_trade_allocation) + _record_risk_diagnostics(report, post_trade_allocation, phase="post_trade") if not post_trade_allocation.get("execution_permitted", False): report["execution_blocked_reason"] = "risk_execution_not_permitted" diff --git a/docs/portfolio-stoploss-observation.md b/docs/portfolio-stoploss-observation.md new file mode 100644 index 00000000..ff89e4d9 --- /dev/null +++ b/docs/portfolio-stoploss-observation.md @@ -0,0 +1,62 @@ +# Portfolio stoploss observation (report only) + +The cycle report adds `diagnostics.portfolio_stoploss.pre_trade` and +`post_trade`. These are observations only. They do not authorize execution, +change allocations, add stop orders, send notifications, change the cycle +status, or repair the portfolio stoploss gap described in issue #369. + +The observation concerns the legacy portfolio unrealized-PnL stoploss rule +(-20%), not every stop mechanism. Existing strategy-level held-trend stops, +strategy stop diagnostics, daily circuit breakers, and MEMBER gate decisions +remain unchanged. + +## Meaning + +- `evaluation_status: not_evaluated` with + `reason_code: legacy_rule_not_in_reviewed_gate` means the reviewed dependency + path uses `assess_with_evidence`, without applying the legacy portfolio rule. + This is not a passed risk check, even if cost data are complete and the + MEMBER assessment says APPROVE. +- `evaluation_status: unknown` with + `reason_code: gate_evaluation_evidence_unavailable` means the installed + revisions, catalog entrypoint wrapper, or MEMBER assessment contract could + not be verified against the reviewed path. No enforcement conclusion follows. +- `evaluation_status: not_evaluated` with + `reason_code: evaluation_not_reached` means that cycle phase has not reached + evaluation, for example after an early failure or pre-trade rejection. +- `pnl_input_status` separately describes average-cost coverage: `missing`, + `partial`, `complete`, `empty`, `invalid`, or `unknown`. Complete coverage is + an input observation, not evidence a stoploss was evaluated. Empty positions + do not establish complete cost coverage. No PnL is computed by this helper. + +`position_count` and `cost_covered_position_count` count positions in the +strategy-member snapshot after managed-symbol filtering. They are not +whole-account coverage, asset-value percentages, balances, or an account PnL. +Fuel/nonmember assets may be excluded. Unknown/invalid input counts are null. +The current snapshot builder supplies no `average_cost`, so a valid nonempty +snapshot currently reports missing cost inputs. No account identifiers, symbols, +monetary values, PnL values, or arbitrary runtime metadata are added to this +observation. Both metadata-to-allocation and allocation-to-report boundaries +use a strict projection. + +## Reviewed scope and future upgrades + +The reviewed path is CryptoStrategies +`7fcca8ee0280b3e66245eb673caadda8b8b3b9a3` and QuantPlatformKit +`155d74a615ccdebf7223787939c2dc7151d75efd`, with the catalog crypto entrypoint +and `qsl.risk_gate_assessment.v2` MEMBER assessment. The common wrapper passes +no risk-control state; the gate's extra stop/drawdown mandate fields apply to +specific TQQQ schemas, not this crypto path, and its risk engine uses no +legacy portfolio-stop detector. Review this conclusion again when dependencies +or entrypoint/gate contracts change. Do not infer it from a dependency version +number, from a MEMBER approval, or from missing risk flags. + +Installed Git revision provenance is required for the reviewed-path label. +Editable installs without that provenance conservatively report unknown. CI +uses editable dependency overlays, so tests explicitly cover both known and +unknown provenance rather than pretending an editable checkout is verified. + +Synthetic tests cover cost coverage, invalid/nonfinite input, strict redaction, +unknown versions/contracts, independent phases, early exit, and unchanged +strategy decisions, cycle action callbacks, notifications, and report status. +No live account, order, notification, deployment, or trading workflow is used. diff --git a/main.py b/main.py index 3b35661e..de6ad4f9 100644 --- a/main.py +++ b/main.py @@ -100,6 +100,7 @@ map_strategy_decision_to_allocation as map_decision_to_allocation, map_strategy_decision_to_rotation_plan as map_decision_to_rotation_plan, ) +from portfolio_stoploss_observation import public_stoploss_observation from strategy_runtime import load_strategy_runtime from run_context import ( RunContext, @@ -1073,6 +1074,9 @@ def _compute_portfolio_allocation(runtime, runtime_trend_universe, balances, pri assessment = evaluation.decision.diagnostics.get("member_risk_assessment") if isinstance(assessment, Mapping): allocation["risk_assessment"] = dict(assessment) + allocation["portfolio_stoploss_observation"] = public_stoploss_observation( + evaluation.metadata.get("portfolio_stoploss_observation") + ) allocation["risk_flags"] = tuple(str(flag) for flag in evaluation.decision.risk_flags) return allocation diff --git a/portfolio_stoploss_observation.py b/portfolio_stoploss_observation.py new file mode 100644 index 00000000..45d110a0 --- /dev/null +++ b/portfolio_stoploss_observation.py @@ -0,0 +1,142 @@ +"""Report-only observation of the legacy portfolio stoploss, never an order gate. + +Cost coverage is counted over the strategy-member snapshot, not account value. +A MEMBER approval is not evidence that the legacy -20% PnL rule ran. Reviewed +pinned dependencies use assess_with_evidence without that legacy detector; +unknown dependencies or assessment contracts must not inherit that conclusion. +""" +from __future__ import annotations + +import json +import math +from collections.abc import Mapping +from importlib.metadata import PackageNotFoundError, distribution + + +_REVIEWED_REVISIONS = { + "quant-platform-kit": "155d74a615ccdebf7223787939c2dc7151d75efd", + "crypto-strategies": "7fcca8ee0280b3e66245eb673caadda8b8b3b9a3", +} +_PROFILES = frozenset({"crypto_live_pool_rotation", "crypto_equity_combo", + "crypto_btc_dca", "crypto_trend_rotation"}) + + +def _reviewed_member_gate(entrypoint, assessment) -> bool: + if not isinstance(assessment, Mapping) or ( + assessment.get("contract_version") != "qsl.risk_gate_assessment.v2" + or assessment.get("scope") != "MEMBER" + or assessment.get("outcome") not in ("APPROVE", "REJECT") + ): + return False + try: + for package, revision in _REVIEWED_REVISIONS.items(): + provenance = json.loads(distribution(package).read_text("direct_url.json") or "{}") + if provenance.get("vcs_info", {}).get("commit_id") != revision: + return False + # The runtime loader wraps a catalog CallableStrategyEntrypoint once. + # Custom entrypoints and changed wrapper contracts remain unknown. + from crypto_strategies import entrypoints + from quant_platform_kit.common.strategy_contracts import CallableStrategyEntrypoint + + profile = entrypoint.manifest.profile + if profile not in _PROFILES or type(entrypoint) is not CallableStrategyEntrypoint: + return False + inner = getattr(entrypoint._evaluate, "__self__", None) + return (type(inner) is CallableStrategyEntrypoint + and inner._evaluate is getattr(entrypoints, f"evaluate_{profile}", None)) + except (PackageNotFoundError, ValueError, TypeError, AttributeError, OSError): + return False + + +def _finite_number(value): + if isinstance(value, bool) or not isinstance(value, (int, float)): + return None + try: + return float(value) if math.isfinite(value) else None + except (OverflowError, ValueError): + return None + + +def _input_coverage(snapshot): + metadata = getattr(snapshot, "metadata", None) + positions = getattr(snapshot, "positions", None) + equity = _finite_number(getattr(snapshot, "total_equity", None)) + if (not isinstance(metadata, Mapping) or metadata.get("snapshot_validation_errors") + or not isinstance(positions, (tuple, list)) or equity is None or equity <= 0): + return "invalid", None, None + covered = 0 + for position in positions: + quantity = _finite_number(getattr(position, "quantity", None)) + value = _finite_number(getattr(position, "market_value", None)) + raw_cost = getattr(position, "average_cost", None) + cost = _finite_number(raw_cost) + if (quantity is None or quantity <= 0 or value is None or value < 0 + or (raw_cost is not None and (cost is None or cost <= 0))): + return "invalid", None, None + covered += cost is not None + count = len(positions) + status = ("empty" if not count else "complete" if covered == count + else "partial" if covered else "missing") + return status, count, covered + + +def _unknown_observation(): + return { + "evaluation_status": "unknown", + "reason_code": "gate_evaluation_evidence_unavailable", + "pnl_input_status": "unknown", + "coverage_basis": "strategy_member_position_count", + "position_count": None, + "cost_covered_position_count": None, + } + + +def unevaluated_stoploss_observation(): + """A cycle phase that has not reached strategy evaluation.""" + result = _unknown_observation() + result.update(evaluation_status="not_evaluated", reason_code="evaluation_not_reached") + return result + + +def observe_portfolio_stoploss(snapshot, *, entrypoint, assessment): + """Return only bounded labels/counts; do not evaluate PnL or change decisions.""" + result = _unknown_observation() + if _reviewed_member_gate(entrypoint, assessment): + result.update(evaluation_status="not_evaluated", reason_code="legacy_rule_not_in_reviewed_gate") + status, count, covered = _input_coverage(snapshot) + result.update(pnl_input_status=status, position_count=count, cost_covered_position_count=covered) + return result + + +def public_stoploss_observation(value): + """Whitelist at report boundaries; malformed metadata cannot affect trading.""" + try: + return _project_stoploss_observation(value) + except Exception: # noqa: BLE001 - no raw errors or reporting failures escape + return _unknown_observation() + + +def _project_stoploss_observation(value): + fallback = _unknown_observation() + if not isinstance(value, Mapping): + return fallback + projected = {key: value.get(key) for key in fallback} + if any(type(projected[key]) is not str for key in ( + "evaluation_status", "reason_code", "pnl_input_status", "coverage_basis" + )): + return fallback + if ((projected["evaluation_status"], projected["reason_code"]) not in { + ("unknown", "gate_evaluation_evidence_unavailable"), + ("not_evaluated", "legacy_rule_not_in_reviewed_gate"), + ("not_evaluated", "evaluation_not_reached"), + } or projected["coverage_basis"] != fallback["coverage_basis"] + or projected["pnl_input_status"] not in {"unknown", "invalid", "empty", "missing", "partial", "complete"}): + return fallback + count, covered = projected["position_count"], projected["cost_covered_position_count"] + if projected["pnl_input_status"] in {"unknown", "invalid"}: + return projected if count is None and covered is None else fallback + if type(count) is not int or type(covered) is not int or not 0 <= covered <= count: + return fallback + expected = ("empty" if not count else "complete" if covered == count + else "partial" if covered else "missing") + return projected if projected["pnl_input_status"] == expected else fallback diff --git a/strategy_runtime.py b/strategy_runtime.py index a41c9c74..06c3f173 100644 --- a/strategy_runtime.py +++ b/strategy_runtime.py @@ -24,6 +24,7 @@ bind_live_risk_authority, config_sha256, ) +from portfolio_stoploss_observation import observe_portfolio_stoploss, public_stoploss_observation from strategy_loader import ( load_research_strategy_entrypoint_for_profile, load_strategy_entrypoint_for_profile, @@ -459,10 +460,19 @@ def invalid_snapshot_number(value: Any) -> float | None: artifacts=artifacts, ) decision = self.entrypoint.evaluate(ctx) + try: + stoploss_observation = observe_portfolio_stoploss( + portfolio_snapshot, + entrypoint=self.entrypoint, + assessment=decision.diagnostics.get("member_risk_assessment"), + ) + except Exception: # noqa: BLE001 - report-only failure must not change execution + stoploss_observation = public_stoploss_observation(None) return StrategyEvaluationResult( decision=decision, account_metrics=dict(account_metrics), metadata={ + "portfolio_stoploss_observation": stoploss_observation, "strategy_profile": self.profile, "strategy_display_name": resolve_research_strategy_metadata( self.profile, diff --git a/tests/test_cycle_service.py b/tests/test_cycle_service.py index ac2570c3..8b77dcfc 100644 --- a/tests/test_cycle_service.py +++ b/tests/test_cycle_service.py @@ -9,6 +9,7 @@ from types import SimpleNamespace from unittest.mock import Mock, patch +from portfolio_stoploss_observation import unevaluated_stoploss_observation from application.cycle_service import ( _write_lifecycle_export, execute_strategy_cycle, @@ -48,11 +49,14 @@ def _run_funds_cycle( rebase_fn=None, reset_fn=None, trend_pool_resolution=None, + stoploss_observations=None, + notifications=None, ): events = [] state = {} if state is None else state post_execution_permitted = execution_permitted if post_execution_permitted is None else post_execution_permitted allocation_permissions = [execution_permitted, post_execution_permitted] + stoploss_observations = list(stoploss_observations or [None, None]) def manage_earn(*_args, **_kwargs): events.append("earn") @@ -120,6 +124,7 @@ def note(event, value): ), top_up_bnb_fuel=lambda *_args: events.append("fuel") or (185.0, 15.0, "ready"), compute_portfolio_allocation=lambda *_args: { + "portfolio_stoploss_observation": stoploss_observations.pop(0), "total_equity": 200.0, "trend_val": 0.0, "dca_val": 0.0, @@ -142,12 +147,44 @@ def note(event, value): maybe_send_periodic_btc_status_report=send_periodic_status, runtime_set_trade_state=lambda *_args, **_kwargs: events.append("state_write"), append_report_error=lambda *_args, **_kwargs: None, - runtime_notify=lambda *_args, **_kwargs: None, + runtime_notify=lambda *args, **kwargs: notifications.append((args, kwargs)) if notifications is not None else None, translate_fn=lambda value, **_kwargs: value, traceback_module=SimpleNamespace(), ) return report, events + def test_stoploss_reporting_preserves_execution_and_notifications(self): + from portfolio_stoploss_observation import observe_portfolio_stoploss + from quant_platform_kit import PortfolioSnapshot, Position + + now = datetime(2026, 10, 1, tzinfo=timezone.utc) + observations = [] + with patch("portfolio_stoploss_observation._reviewed_member_gate", return_value=True): + for costs in ((None, None), (1.0, None), (1.0, 2.0)): + snapshot = PortfolioSnapshot(as_of=now, total_equity=1000.0, positions=tuple( + Position("SYNTHETIC", 1.0, 3.0, cost) for cost in costs + )) + observations.append(observe_portfolio_stoploss(snapshot, entrypoint=None, assessment={})) + baseline_notifications = [] + baseline, baseline_events = self._run_funds_cycle(True, notifications=baseline_notifications) + baseline.pop("diagnostics") + for observation in observations: + notifications = [] + report, events = self._run_funds_cycle( + True, stoploss_observations=[observation, observations[-1]], notifications=notifications, + ) + phases = report.pop("diagnostics")["portfolio_stoploss"] + self.assertEqual(phases["pre_trade"], observation) + self.assertEqual(phases["post_trade"], observations[-1]) + self.assertEqual(report, baseline) + self.assertEqual(events, baseline_events) + self.assertEqual(notifications, baseline_notifications) + rejected, events = self._run_funds_cycle(False, stoploss_observations=[observations[0], observations[-1]]) + self.assertEqual(rejected["diagnostics"]["portfolio_stoploss"]["pre_trade"], observations[0]) + self.assertEqual(rejected["diagnostics"]["portfolio_stoploss"]["post_trade"], unevaluated_stoploss_observation()) + self.assertNotIn("trend", events) + self.assertNotIn("dca", events) + def test_cycle_report_preserves_original_upstream_pool_symbols(self): report, _events = self._run_funds_cycle( False, @@ -1320,6 +1357,10 @@ def test_owner_failure_is_safe_and_does_not_reach_client_or_release_lock(self): self.assertEqual(report["status"], "error") self.assertEqual(report["diagnostics"], { "existing": "preserved", + "portfolio_stoploss": { + phase: unevaluated_stoploss_observation() + for phase in ("pre_trade", "post_trade") + }, "cycle_failure": {"stage": "state_owner_claim", "error_type": expected_type}, }) callbacks["ensure_runtime_client"].assert_not_called() diff --git a/tests/test_portfolio_stoploss_observation.py b/tests/test_portfolio_stoploss_observation.py new file mode 100644 index 00000000..2cf07152 --- /dev/null +++ b/tests/test_portfolio_stoploss_observation.py @@ -0,0 +1,171 @@ +"""Synthetic-only reporting tests. No exchange or notification calls.""" +import ast +import json +from pathlib import Path +from collections.abc import Mapping +import unittest +from datetime import datetime, timezone +from types import SimpleNamespace +from unittest.mock import patch + +from quant_platform_kit import PortfolioSnapshot, Position + +from portfolio_stoploss_observation import observe_portfolio_stoploss, public_stoploss_observation + + +class PortfolioStoplossObservationTests(unittest.TestCase): + def snapshot(self, positions=(), **kwargs): + return PortfolioSnapshot(as_of=datetime.now(timezone.utc), total_equity=1000.0, + positions=positions, **kwargs) + + def observe(self, snapshot, known=True, assessment=None): + if assessment is None: + assessment = {"scope": "MEMBER", "outcome": "PASS"} + with patch("portfolio_stoploss_observation._reviewed_member_gate", return_value=known): + return observe_portfolio_stoploss(snapshot, entrypoint=None, assessment=assessment) + + def test_cost_coverage_is_position_count_and_never_a_pass(self): + for costs, status, count in [((None, None), "missing", 0), + ((10.0, None), "partial", 1), + ((10.0, 20.0), "complete", 2)]: + with self.subTest(costs=costs): + positions = tuple(Position("PRIVATE", 1.0, 25.0, cost) for cost in costs) + result = self.observe(self.snapshot(positions)) + self.assertEqual(result["pnl_input_status"], status) + self.assertEqual(result["cost_covered_position_count"], count) + self.assertEqual(result["position_count"], 2) + self.assertEqual(result["evaluation_status"], "not_evaluated") + self.assertEqual(result["coverage_basis"], "strategy_member_position_count") + self.assertNotIn("passed", json.dumps(result)) + + def test_empty_is_not_full_cost_coverage(self): + result = self.observe(self.snapshot()) + self.assertEqual(result["pnl_input_status"], "empty") + self.assertEqual(result["position_count"], 0) + + def test_invalid_snapshot_and_nonfinite_values_are_unknown(self): + snapshots = [None, SimpleNamespace(positions="bad"), + self.snapshot(metadata={"snapshot_validation_errors": ("PRIVATE",)})] + for field in ("quantity", "market_value", "average_cost"): + for bad in (True, float("nan"), float("inf"), "PRIVATE"): + values = dict(symbol="PRIVATE", quantity=1.0, market_value=25.0, average_cost=10.0) + values[field] = bad + snapshots.append(self.snapshot((Position(**values),))) + for snapshot in snapshots: + with self.subTest(snapshot=snapshot): + result = self.observe(snapshot) + self.assertEqual(result["pnl_input_status"], "invalid") + self.assertIsNone(result["cost_covered_position_count"]) + self.assertNotIn("PRIVATE", json.dumps(result)) + + def test_unreviewed_gate_is_unknown_even_with_complete_costs(self): + result = self.observe(self.snapshot((Position("PRIVATE", 1, 1, 1),)), known=False) + self.assertEqual(result["evaluation_status"], "unknown") + self.assertEqual(result["reason_code"], "gate_evaluation_evidence_unavailable") + + def test_public_projection_contains_unexpected_mapping_errors(self): + class RaisingMapping(Mapping): + def __iter__(self): + return iter(()) + def __len__(self): + return 0 + def __getitem__(self, key): + raise RuntimeError("PRIVATE token") + result = public_stoploss_observation(RaisingMapping()) + self.assertEqual(result["evaluation_status"], "unknown") + self.assertNotIn("PRIVATE", json.dumps(result)) + + def test_public_projection_is_strict_and_fail_closed(self): + good = self.observe(self.snapshot()) + dirty = {**good, "account_id": "PRIVATE", "pnl": -0.42, "token": "SECRET"} + self.assertEqual(public_stoploss_observation(dirty), good) + for dirty in [None, {**good, "reason_code": "SECRET"}, + {**good, "evaluation_status": "passed"}, + {**good, "position_count": float("nan")}, + {**good, "reason_code": []}]: + result = public_stoploss_observation(dirty) + self.assertEqual(result["evaluation_status"], "unknown") + self.assertNotIn("SECRET", json.dumps(result)) + + +class ReviewedGateTests(unittest.TestCase): + def setUp(self): + from strategy_runtime import load_research_only_strategy_runtime + self.runtime = load_research_only_strategy_runtime("crypto_live_pool_rotation") + self.assessment = {"contract_version": "qsl.risk_gate_assessment.v2", + "scope": "MEMBER", "outcome": "APPROVE"} + + def test_review_requires_exact_installed_revisions_and_gate_contract(self): + from portfolio_stoploss_observation import _reviewed_member_gate, _REVIEWED_REVISIONS + + def installed(package): + return SimpleNamespace(read_text=lambda _: json.dumps({ + "vcs_info": {"commit_id": _REVIEWED_REVISIONS[package]}})) + + with patch("portfolio_stoploss_observation.distribution", side_effect=installed): + self.assertTrue(_reviewed_member_gate(self.runtime.entrypoint, self.assessment)) + for assessment in (None, {}, {**self.assessment, "scope": "ACCOUNT"}, + {**self.assessment, "contract_version": "future"}, + {**self.assessment, "outcome": []}): + self.assertFalse(_reviewed_member_gate(self.runtime.entrypoint, assessment)) + custom = SimpleNamespace(manifest=self.runtime.entrypoint.manifest) + self.assertFalse(_reviewed_member_gate(custom, self.assessment)) + for provenance in ({}, {"dir_info": {"editable": True}}, + {"vcs_info": {"commit_id": "future"}}, []): + with patch("portfolio_stoploss_observation.distribution", return_value=SimpleNamespace( + read_text=lambda _: json.dumps(provenance) + )): + self.assertFalse(_reviewed_member_gate(self.runtime.entrypoint, self.assessment)) + + def test_phase_projection_preserves_existing_risk_and_redacts_metadata(self): + from application.cycle_service import _record_risk_diagnostics + report = {"status": "ok", "diagnostics": {"existing": True}} + observation = PortfolioStoplossObservationTests().observe( + PortfolioStoplossObservationTests().snapshot()) + allocation = {"portfolio_stoploss_observation": {**observation, "token": "SECRET"}, + "risk_assessment": self.assessment, "risk_flags": ("original",)} + _record_risk_diagnostics(report, allocation, phase="pre_trade") + _record_risk_diagnostics(report, {}, phase="post_trade") + self.assertEqual(report["diagnostics"]["portfolio_stoploss"]["pre_trade"], observation) + self.assertEqual(report["diagnostics"]["portfolio_stoploss"]["post_trade"]["evaluation_status"], "unknown") + self.assertEqual(report["risk_assessment"], self.assessment) + self.assertEqual(report["risk_outcome"], "APPROVE") + self.assertEqual(report["risk_flags"], ["original"]) + self.assertEqual(report["status"], "ok") + self.assertTrue(report["diagnostics"]["existing"]) + self.assertNotIn("SECRET", json.dumps(report)) + + +class AllocationProjectionTests(unittest.TestCase): + def test_main_allocation_whitelists_observation_without_changing_decision(self): + # Compile the actual adapter function without starting main's runtime imports. + source = ast.parse((Path(__file__).resolve().parents[1] / "main.py").read_text()) + function = next(node for node in source.body if isinstance(node, ast.FunctionDef) + and node.name == "_compute_portfolio_allocation") + observation = PortfolioStoplossObservationTests().observe( + PortfolioStoplossObservationTests().snapshot()) + decision = SimpleNamespace(diagnostics={"member_risk_assessment": {"outcome": "APPROVE"}}, + risk_flags=("original",)) + metrics = {"original_metric": 1} + evaluation = SimpleNamespace(decision=decision, account_metrics=metrics, metadata={ + "portfolio_stoploss_observation": {**observation, "account_id": "PRIVATE", "pnl": 12}, + "secret": "PRIVATE", + }) + mapped = [] + def mapper(value, *, account_metrics): + mapped.append((value, account_metrics)) + return {"execution_permitted": True, "original": True} + namespace = {"Mapping": Mapping, "public_stoploss_observation": public_stoploss_observation, + "_resolve_strategy_evaluation": lambda *args, **kwargs: evaluation, + "map_decision_to_allocation": mapper} + exec(compile(ast.Module(body=[function], type_ignores=[]), "main.py", "exec"), namespace) + result = namespace["_compute_portfolio_allocation"](*([None] * 9)) + self.assertEqual(mapped, [(decision, metrics)]) + self.assertEqual(result, {"execution_permitted": True, "original": True, + "risk_assessment": {"outcome": "APPROVE"}, "risk_flags": ("original",), + "portfolio_stoploss_observation": observation}) + self.assertNotIn("PRIVATE", json.dumps(result)) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_strategy_runtime.py b/tests/test_strategy_runtime.py index a46179eb..54b37a0a 100644 --- a/tests/test_strategy_runtime.py +++ b/tests/test_strategy_runtime.py @@ -90,6 +90,38 @@ def _evaluate_minimal_runtime( class StrategyRuntimeTests(unittest.TestCase): + def test_stoploss_observation_does_not_mutate_strategy_decision(self): + import strategy_runtime as module + from quant_platform_kit.common.strategy_contracts import StrategyDecision + + runtime = module.load_research_only_strategy_runtime("crypto_live_pool_rotation") + decision = StrategyDecision(diagnostics={ + "member_risk_assessment": {"scope": "MEMBER", "outcome": "APPROVE"}, + "strategy_stop_evaluation": {"original": "preserved"}, + }) + now = datetime(2026, 10, 1, tzinfo=timezone.utc) + candidate = _synthetic_candidate_identity() + with patch.object(type(runtime.entrypoint), "evaluate", return_value=decision): + evaluation = _evaluate_minimal_runtime(runtime, now=now, + mandate=_synthetic_mandate(now, candidate), candidate=candidate) + self.assertIs(evaluation.decision, decision) + self.assertEqual(decision.diagnostics["strategy_stop_evaluation"], {"original": "preserved"}) + observation = evaluation.metadata["portfolio_stoploss_observation"] + self.assertEqual(observation["position_count"], 1) + self.assertEqual(observation["pnl_input_status"], "missing") + self.assertEqual(observation["evaluation_status"], "unknown") + self.assertNotIn("portfolio_stoploss_observation", decision.diagnostics) + with patch.object(type(runtime.entrypoint), "evaluate", return_value=decision), patch.object( + module, "observe_portfolio_stoploss", side_effect=RuntimeError("PRIVATE token") + ): + failed_observation = _evaluate_minimal_runtime(runtime, now=now, + mandate=_synthetic_mandate(now, candidate), candidate=candidate) + self.assertIs(failed_observation.decision, decision) + self.assertEqual(failed_observation.account_metrics, evaluation.account_metrics) + self.assertEqual(failed_observation.metadata["portfolio_stoploss_observation"]["pnl_input_status"], "unknown") + self.assertNotIn("PRIVATE", str(failed_observation.metadata)) + + def test_build_portfolio_snapshot_values_all_non_usdt_risk_assets(self): import strategy_runtime as strategy_runtime_module