report: add truthful portfolio stoploss observations

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"

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.

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
 

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

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,

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()