07.4.4.1.10.1 Semantic Diagnostic Snapshot Builder
This commit is contained in:
1
app/src/trading/diagnostics/__init__.py
Normal file
1
app/src/trading/diagnostics/__init__.py
Normal file
@@ -0,0 +1 @@
|
||||
from __future__ import annotations
|
||||
238
app/src/trading/diagnostics/semantic_runtime.py
Normal file
238
app/src/trading/diagnostics/semantic_runtime.py
Normal file
@@ -0,0 +1,238 @@
|
||||
# app/src/trading/diagnostics/semantic_runtime.py
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import time
|
||||
from typing import Any
|
||||
|
||||
from src.trading.auto.state import AutoTradeState
|
||||
|
||||
|
||||
class SemanticRuntimeDiagnostics:
|
||||
def build(self, state: AutoTradeState) -> dict[str, Any]:
|
||||
now = time.monotonic()
|
||||
|
||||
return {
|
||||
"snapshot_type": "SEMANTIC_RUNTIME_DIAGNOSTIC",
|
||||
"snapshot_version": "07.4.4.1.10.1",
|
||||
"built_at_monotonic": now,
|
||||
"status": self._status_section(state),
|
||||
"signal": self._signal_section(state, now),
|
||||
"market": self._market_section(state, now),
|
||||
"momentum": self._momentum_section(state),
|
||||
"execution": self._execution_section(state),
|
||||
"adaptive_size": self._adaptive_size_section(state),
|
||||
"position": self._position_section(state),
|
||||
"runtime_health": self._runtime_health_section(state, now),
|
||||
"summary": self._summary_section(state),
|
||||
}
|
||||
|
||||
def _status_section(self, state: AutoTradeState) -> dict[str, Any]:
|
||||
return {
|
||||
"status": state.status,
|
||||
"symbol": state.symbol,
|
||||
"strategy": state.strategy,
|
||||
"last_check_at": state.last_check_at,
|
||||
"is_configured": self._is_configured(state),
|
||||
}
|
||||
|
||||
def _signal_section(
|
||||
self,
|
||||
state: AutoTradeState,
|
||||
now: float,
|
||||
) -> dict[str, Any]:
|
||||
signal_age_seconds = self._age_seconds(
|
||||
started_at=state.signal_started_at,
|
||||
now=now,
|
||||
)
|
||||
|
||||
return {
|
||||
"signal": state.last_signal,
|
||||
"confidence": state.last_signal_confidence,
|
||||
"reason": state.last_signal_reason,
|
||||
"repeat_count": state.last_signal_repeat_count,
|
||||
"started_at": state.signal_started_at,
|
||||
"updated_at": state.signal_updated_at,
|
||||
"age_seconds": signal_age_seconds,
|
||||
"decision_status": state.decision_status,
|
||||
"decision_reason": state.decision_reason,
|
||||
"is_confirmed": state.is_signal_confirmed,
|
||||
"is_ready": state.is_signal_ready,
|
||||
"confirmation_seconds": state.signal_confirmation_seconds,
|
||||
"confirmation_required_seconds": state.signal_confirmation_required_seconds,
|
||||
"confirmation_missing_repeats": state.signal_confirmation_missing_repeats,
|
||||
"confirmation_progress": state.signal_confirmation_progress,
|
||||
"confirmation_reason": state.signal_confirmation_reason,
|
||||
}
|
||||
|
||||
def _market_section(
|
||||
self,
|
||||
state: AutoTradeState,
|
||||
now: float,
|
||||
) -> dict[str, Any]:
|
||||
market_age_seconds = self._age_seconds(
|
||||
started_at=state.market_analysis_updated_at,
|
||||
now=now,
|
||||
)
|
||||
|
||||
return {
|
||||
"state": state.market_state,
|
||||
"trend": state.market_trend,
|
||||
"volatility": state.market_volatility,
|
||||
"trend_strength": state.market_trend_strength,
|
||||
"trend_quality": state.market_trend_quality,
|
||||
"phase": state.market_phase,
|
||||
"phase_direction": state.market_phase_direction,
|
||||
"interval": state.market_analysis_interval,
|
||||
"reason": state.market_analysis_reason,
|
||||
"updated_at": state.market_analysis_updated_at,
|
||||
"age_seconds": market_age_seconds,
|
||||
"entry_block_reason": state.entry_block_reason,
|
||||
"entry_block_message": state.entry_block_message,
|
||||
}
|
||||
|
||||
def _momentum_section(self, state: AutoTradeState) -> dict[str, Any]:
|
||||
return {
|
||||
"state": state.momentum_state,
|
||||
"direction": state.momentum_direction,
|
||||
"change_percent": state.momentum_change_percent,
|
||||
"strength": state.momentum_strength,
|
||||
"breakout_level": state.breakout_level,
|
||||
"breakout_distance_percent": state.breakout_distance_percent,
|
||||
"breakout_reason": state.breakout_reason,
|
||||
"is_breakout": state.momentum_state in {
|
||||
"BREAKOUT_UP",
|
||||
"BREAKOUT_DOWN",
|
||||
},
|
||||
"is_momentum": state.momentum_state in {
|
||||
"MOMENTUM_UP",
|
||||
"MOMENTUM_DOWN",
|
||||
"BREAKOUT_UP",
|
||||
"BREAKOUT_DOWN",
|
||||
},
|
||||
}
|
||||
|
||||
def _execution_section(self, state: AutoTradeState) -> dict[str, Any]:
|
||||
return {
|
||||
"quality": state.execution_quality,
|
||||
"quality_reason": state.execution_quality_reason,
|
||||
"quality_message": state.execution_quality_message,
|
||||
"semantic_status": state.execution_semantic_status,
|
||||
"semantic_message": state.execution_semantic_message,
|
||||
"semantic_reason": state.execution_semantic_reason,
|
||||
"confidence_score": state.execution_confidence_score,
|
||||
"confidence_level": state.execution_confidence_level,
|
||||
"confidence_required_score": state.execution_confidence_required_score,
|
||||
"confidence_reason": state.execution_confidence_reason,
|
||||
"confidence_factors": state.execution_confidence_factors,
|
||||
"block_reason": state.execution_block_reason,
|
||||
"snapshot_age_seconds": state.snapshot_age_seconds,
|
||||
"spread_percent": state.spread_percent,
|
||||
"market_runtime_degraded": state.market_runtime_degraded,
|
||||
}
|
||||
|
||||
def _adaptive_size_section(self, state: AutoTradeState) -> dict[str, Any]:
|
||||
return {
|
||||
"base_size": state.adaptive_size_base,
|
||||
"final_size": state.adaptive_size_final,
|
||||
"multiplier": state.adaptive_size_multiplier,
|
||||
"reason": state.adaptive_size_reason,
|
||||
"factors": state.adaptive_size_factors,
|
||||
"effective_risk_percent": state.effective_risk_percent,
|
||||
"effective_target_risk_usd": state.effective_target_risk_usd,
|
||||
"size_adjustment_reason": state.execution_size_adjustment_reason,
|
||||
}
|
||||
|
||||
def _position_section(self, state: AutoTradeState) -> dict[str, Any]:
|
||||
return {
|
||||
"side": state.position_side,
|
||||
"entry_price": state.entry_price,
|
||||
"size": state.position_size,
|
||||
"unrealized_pnl_usd": state.unrealized_pnl_usd,
|
||||
"realized_pnl_usd": state.realized_pnl_usd,
|
||||
"last_execution_action": state.last_execution_action,
|
||||
"last_execution_reason": state.last_execution_reason,
|
||||
"last_flip_block_reason": state.last_flip_block_reason,
|
||||
"last_flip_at": state.last_flip_at,
|
||||
}
|
||||
|
||||
def _runtime_health_section(
|
||||
self,
|
||||
state: AutoTradeState,
|
||||
now: float,
|
||||
) -> dict[str, Any]:
|
||||
signal_age_seconds = self._age_seconds(
|
||||
started_at=state.signal_updated_at,
|
||||
now=now,
|
||||
)
|
||||
market_age_seconds = self._age_seconds(
|
||||
started_at=state.market_analysis_updated_at,
|
||||
now=now,
|
||||
)
|
||||
|
||||
return {
|
||||
"runtime_expired_reason": state.runtime_expired_reason,
|
||||
"runtime_expired_message": state.runtime_expired_message,
|
||||
"signal_age_seconds": signal_age_seconds,
|
||||
"market_age_seconds": market_age_seconds,
|
||||
"has_market_data": state.market_state is not None,
|
||||
"has_execution_quality": state.execution_quality is not None,
|
||||
"has_signal": state.last_signal is not None,
|
||||
"has_momentum_data": state.momentum_state is not None,
|
||||
"is_runtime_degraded": bool(state.market_runtime_degraded),
|
||||
}
|
||||
|
||||
def _summary_section(self, state: AutoTradeState) -> dict[str, Any]:
|
||||
blockers = []
|
||||
|
||||
if state.entry_block_message:
|
||||
blockers.append(state.entry_block_message)
|
||||
|
||||
if state.execution_quality == "BLOCKED":
|
||||
blockers.append(state.execution_quality_message or "execution blocked")
|
||||
|
||||
if state.decision_status == "BLOCKED":
|
||||
blockers.append(state.decision_reason or "decision blocked")
|
||||
|
||||
return {
|
||||
"mode": state.status,
|
||||
"signal": state.last_signal,
|
||||
"market": state.market_state,
|
||||
"phase": state.market_phase,
|
||||
"momentum": state.momentum_state,
|
||||
"execution": state.execution_semantic_status,
|
||||
"position": state.position_side,
|
||||
"is_trade_candidate": state.last_signal in {"BUY", "SELL"},
|
||||
"is_ready": bool(state.is_signal_ready),
|
||||
"is_blocked": bool(blockers),
|
||||
"blockers": blockers,
|
||||
}
|
||||
|
||||
def _is_configured(self, state: AutoTradeState) -> bool:
|
||||
if not state.symbol:
|
||||
return False
|
||||
|
||||
if not state.strategy:
|
||||
return False
|
||||
|
||||
if state.risk_percent is None:
|
||||
return False
|
||||
|
||||
if state.strategy.upper() == "TREND":
|
||||
return (
|
||||
state.stop_loss_percent is not None
|
||||
and state.stop_loss_percent > 0
|
||||
)
|
||||
|
||||
return True
|
||||
|
||||
def _age_seconds(
|
||||
self,
|
||||
*,
|
||||
started_at: float | None,
|
||||
now: float,
|
||||
) -> int | None:
|
||||
if started_at is None:
|
||||
return None
|
||||
|
||||
return max(0, int(now - float(started_at)))
|
||||
Reference in New Issue
Block a user