07.4.4.1.10.2 Human-readable formatter

app/src/trading/diagnostics/formatter.py

@@ -0,0 +1,200 @@
+# app/src/trading/diagnostics/formatter.py
+
+from __future__ import annotations
+
+from typing import Any
+
+
+class SemanticDiagnosticFormatter:
+    def format(self, snapshot: dict[str, Any]) -> str:
+        sections: list[str] = []
+
+        status = snapshot.get("status", {})
+        signal = snapshot.get("signal", {})
+        market = snapshot.get("market", {})
+        momentum = snapshot.get("momentum", {})
+        execution = snapshot.get("execution", {})
+        adaptive = snapshot.get("adaptive_size", {})
+        runtime = snapshot.get("runtime_health", {})
+        summary = snapshot.get("summary", {})
+        position = snapshot.get("position", {})
+
+        sections.extend(
+            [
+                "🧠 Semantic Runtime Diagnostics",
+                "",
+                self._status_block(status),
+                self._signal_block(signal),
+                self._market_block(market),
+                self._momentum_block(momentum),
+                self._execution_block(execution),
+                self._adaptive_block(adaptive),
+                self._position_block(position),
+                self._runtime_block(runtime),
+                self._summary_block(summary),
+            ]
+        )
+
+        return "\n".join(
+            line
+            for line in sections
+            if line is not None
+        ).strip()
+
+    def _status_block(self, data: dict[str, Any]) -> str:
+        return (
+            "📦 Runtime\n"
+            f"• Status: {data.get('status')}\n"
+            f"• Symbol: {data.get('symbol')}\n"
+            f"• Strategy: {data.get('strategy')}\n"
+            f"• Configured: {self._bool(data.get('is_configured'))}\n"
+        )
+
+    def _signal_block(self, data: dict[str, Any]) -> str:
+        return (
+            "📡 Signal\n"
+            f"• Signal: {data.get('signal')}\n"
+            f"• Confidence: {self._float(data.get('confidence'))}\n"
+            f"• Decision: {data.get('decision_status')}\n"
+            f"• Confirmed: {self._bool(data.get('is_confirmed'))}\n"
+            f"• Ready: {self._bool(data.get('is_ready'))}\n"
+            f"• Repeats: {data.get('repeat_count')}\n"
+            f"• Progress: {self._percent(data.get('confirmation_progress'))}\n"
+            f"• Age: {self._seconds(data.get('age_seconds'))}\n"
+            f"• Reason: {data.get('reason')}\n"
+        )
+
+    def _market_block(self, data: dict[str, Any]) -> str:
+        return (
+            "📈 Market\n"
+            f"• State: {data.get('state')}\n"
+            f"• Trend: {data.get('trend')}\n"
+            f"• Volatility: {data.get('volatility')}\n"
+            f"• Strength: {data.get('trend_strength')}\n"
+            f"• Quality: {data.get('trend_quality')}\n"
+            f"• Phase: {data.get('phase')}\n"
+            f"• Phase Direction: {data.get('phase_direction')}\n"
+            f"• Entry Block: {data.get('entry_block_message')}\n"
+            f"• Analysis Age: {self._seconds(data.get('age_seconds'))}\n"
+        )
+
+    def _momentum_block(self, data: dict[str, Any]) -> str:
+        return (
+            "⚡ Momentum / Breakout\n"
+            f"• State: {data.get('state')}\n"
+            f"• Direction: {data.get('direction')}\n"
+            f"• Strength: {self._float(data.get('strength'))}\n"
+            f"• Change %: {self._float(data.get('change_percent'))}\n"
+            f"• Breakout Level: {self._float(data.get('breakout_level'))}\n"
+            f"• Breakout Distance %: "
+            f"{self._float(data.get('breakout_distance_percent'))}\n"
+            f"• Is Breakout: {self._bool(data.get('is_breakout'))}\n"
+            f"• Reason: {data.get('breakout_reason')}\n"
+        )
+
+    def _execution_block(self, data: dict[str, Any]) -> str:
+        return (
+            "🛡 Execution\n"
+            f"• Quality: {data.get('quality')}\n"
+            f"• Semantic Status: {data.get('semantic_status')}\n"
+            f"• Confidence Score: "
+            f"{self._float(data.get('confidence_score'))}\n"
+            f"• Confidence Level: {data.get('confidence_level')}\n"
+            f"• Spread %: {self._float(data.get('spread_percent'))}\n"
+            f"• Snapshot Age: "
+            f"{self._seconds(data.get('snapshot_age_seconds'))}\n"
+            f"• Runtime Degraded: "
+            f"{self._bool(data.get('market_runtime_degraded'))}\n"
+            f"• Reason: {data.get('semantic_reason')}\n"
+        )
+
+    def _adaptive_block(self, data: dict[str, Any]) -> str:
+        return (
+            "🧮 Adaptive Sizing\n"
+            f"• Multiplier: {self._float(data.get('multiplier'))}\n"
+            f"• Effective Risk %: "
+            f"{self._float(data.get('effective_risk_percent'))}\n"
+            f"• Effective Risk USD: "
+            f"{self._float(data.get('effective_target_risk_usd'))}\n"
+            f"• Reason: {data.get('reason')}\n"
+        )
+
+    def _position_block(self, data: dict[str, Any]) -> str:
+        return (
+            "📌 Position\n"
+            f"• Side: {data.get('side')}\n"
+            f"• Entry: {self._float(data.get('entry_price'))}\n"
+            f"• Size: {self._float(data.get('size'))}\n"
+            f"• Unrealized PnL: "
+            f"{self._float(data.get('unrealized_pnl_usd'))}\n"
+            f"• Realized PnL: "
+            f"{self._float(data.get('realized_pnl_usd'))}\n"
+            f"• Last Action: {data.get('last_execution_action')}\n"
+        )
+
+    def _runtime_block(self, data: dict[str, Any]) -> str:
+        return (
+            "🧬 Runtime Health\n"
+            f"• Runtime Degraded: "
+            f"{self._bool(data.get('is_runtime_degraded'))}\n"
+            f"• Signal Age: "
+            f"{self._seconds(data.get('signal_age_seconds'))}\n"
+            f"• Market Age: "
+            f"{self._seconds(data.get('market_age_seconds'))}\n"
+            f"• Expired Reason: "
+            f"{data.get('runtime_expired_reason')}\n"
+            f"• Has Market Data: "
+            f"{self._bool(data.get('has_market_data'))}\n"
+            f"• Has Momentum Data: "
+            f"{self._bool(data.get('has_momentum_data'))}\n"
+        )
+
+    def _summary_block(self, data: dict[str, Any]) -> str:
+        blockers = data.get("blockers") or []
+
+        if blockers:
+            blockers_text = ", ".join(str(item) for item in blockers)
+        else:
+            blockers_text = "none"
+
+        return (
+            "🧾 Summary\n"
+            f"• Market: {data.get('market')}\n"
+            f"• Phase: {data.get('phase')}\n"
+            f"• Momentum: {data.get('momentum')}\n"
+            f"• Execution: {data.get('execution')}\n"
+            f"• Position: {data.get('position')}\n"
+            f"• Ready: {self._bool(data.get('is_ready'))}\n"
+            f"• Blocked: {self._bool(data.get('is_blocked'))}\n"
+            f"• Blockers: {blockers_text}\n"
+        )
+
+    def _bool(self, value: object) -> str:
+        return "YES" if bool(value) else "NO"
+
+    def _float(self, value: object) -> str:
+        if value is None:
+            return "—"
+
+        try:
+            return f"{float(value):.4f}"
+        except Exception:
+            return str(value)
+
+    def _percent(self, value: object) -> str:
+        if value is None:
+            return "—"
+
+        try:
+            return f"{float(value) * 100:.1f}%"
+        except Exception:
+            return str(value)
+
+    def _seconds(self, value: object) -> str:
+        if value is None:
+            return "—"
+
+        try:
+            return f"{int(float(value))}s"
+        except Exception:
+            return str(value)