feat(strategy_runtime): enrich portfolio metadata with unrealized PnL

application/portfolio_risk_diagnostics.py

@@ -0,0 +1,44 @@
+"""Portfolio risk diagnostics with QPK import fallback until pin bumps."""
+
+from __future__ import annotations
+
+from typing import Any
+
+try:
+    from quant_platform_kit.risk.portfolio_diagnostics import extract_portfolio_risk_diagnostics
+except ImportError:  # pragma: no cover - exercised only before QPK pin bump
+
+    def _position_unrealized_pnl(position: Any) -> float | None:
+        quantity = float(getattr(position, "quantity", 0.0) or 0.0)
+        if quantity == 0.0:
+            return 0.0
+        market_value = float(getattr(position, "market_value", 0.0) or 0.0)
+        average_cost = getattr(position, "average_cost", None)
+        if average_cost is None:
+            return None
+        cost_basis = abs(quantity) * float(average_cost)
+        return market_value - cost_basis
+
+    def extract_portfolio_risk_diagnostics(snapshot: Any) -> dict[str, float | int]:
+        diagnostics: dict[str, float | int] = {}
+        total_equity = float(getattr(snapshot, "total_equity", 0.0) or 0.0)
+        metadata = dict(getattr(snapshot, "metadata", None) or {})
+        if metadata.get("unrealized_pnl_pct") is not None:
+            diagnostics["unrealized_pnl_pct"] = float(metadata["unrealized_pnl_pct"])
+        elif total_equity > 0.0:
+            positions = getattr(snapshot, "positions", ()) or ()
+            unrealized = 0.0
+            has_cost_basis = False
+            for position in positions:
+                position_pnl = _position_unrealized_pnl(position)
+                if position_pnl is None:
+                    continue
+                has_cost_basis = True
+                unrealized += position_pnl
+            if has_cost_basis or not positions:
+                diagnostics["unrealized_pnl_pct"] = float(unrealized / total_equity)
+        if metadata.get("consecutive_losses") is not None:
+            diagnostics["consecutive_losses"] = int(metadata["consecutive_losses"])
+        return diagnostics
+
+__all__ = ["extract_portfolio_risk_diagnostics"]

strategy_runtime.py

@@ -444,6 +444,9 @@ def _enrich_portfolio_metadata(
             for position in getattr(portfolio_snapshot, "positions", ()) or ()
             if str(getattr(position, "symbol", "") or "").strip()
         }
+        from application.portfolio_risk_diagnostics import extract_portfolio_risk_diagnostics
+
+        enriched.update(extract_portfolio_risk_diagnostics(portfolio_snapshot))
         return enriched
 
     def _build_market_history_inputs(

tests/test_strategy_runtime.py

@@ -1183,3 +1183,41 @@ def fake_close_loader(_ib, symbol, duration="2 Y", bar_size="1 day"):
     assert close_loader_symbols == [("TQQQ", "10 D", "1 day"), ("BOXX", "10 D", "1 day")]
     assert result.metadata["price_fallback_source"] == "historical_close"
     assert result.metadata["price_fallbacks"] == {"TQQQ": 70.0, "BOXX": 105.0}
+
+
+def test_enrich_portfolio_metadata_includes_unrealized_pnl():
+    from datetime import datetime, timezone
+
+    from quant_platform_kit.common.models import Position
+
+    class FakeEntrypoint:
+        manifest = StrategyManifest(
+            profile="tech_communication_pullback_enhancement",
+            domain="us_equity",
+            display_name="Tech Pullback",
+            description="test",
+            required_inputs=frozenset(),
+        )
+
+        def evaluate(self, ctx):
+            return StrategyDecision(positions=())
+
+    runtime = strategy_runtime_module.LoadedStrategyRuntime(
+        entrypoint=FakeEntrypoint(),
+        runtime_adapter=StrategyRuntimeAdapter(),
+        runtime_settings=_build_runtime_settings(),
+    )
+    snapshot = PortfolioSnapshot(
+        as_of=datetime.now(timezone.utc),
+        total_equity=10_000.0,
+        positions=(
+            Position(symbol="SPY", quantity=10.0, market_value=4_500.0, average_cost=500.0),
+        ),
+        metadata={"consecutive_losses": 2},
+    )
+
+    enriched = runtime._enrich_portfolio_metadata({}, snapshot)
+
+    assert enriched["unrealized_pnl_pct"] == -0.05
+    assert enriched["consecutive_losses"] == 2
+    assert enriched["portfolio_total_equity"] == 10_000.0