mirror of
https://github.com/triqbit/mt5-ai-xauusd-trader.git
synced 2026-11-02 06:56:15 +00:00
feat: 📡 Jules02: Observability improvement — heartbeat and iteration summary (#1377)
* feat: 📡 Jules02: Observability improvement — heartbeat and iteration summary - Enhanced `Monitor` class with new Prometheus metrics: `ITERATION_HEARTBEAT_GAUGE`, `ITERATION_DURATION_HISTOGRAM`, and `MARKET_STABILITY_GAUGE`. - Updated `send_message` to automatically append `trace_id` to Telegram alerts for direct correlation with system logs. - Instrumented the `run_live` loop in `main.py` to emit a structured `iteration_summary` log at the end of every cycle, providing high-fidelity failure attribution and processing latency. - Added `tests/test_iteration_summary.py` to verify telemetry enhancements and alert correlation. - Verified system integrity with 32 passing tests. Co-authored-by: xnessom <255288032+xnessom@users.noreply.github.com> * feat: 📡 Jules02: Observability improvement — heartbeat and iteration summary - Enhanced `Monitor` class with new Prometheus metrics: `ITERATION_HEARTBEAT_GAUGE`, `ITERATION_DURATION_HISTOGRAM`, and `MARKET_STABILITY_GAUGE`. - Updated `send_message` to automatically append `trace_id` snippets to Telegram alerts for direct log correlation. - Instrumented the `run_live` loop in `main.py` to emit a structured `iteration_summary` log at the end of every cycle (success or failure), providing high-fidelity failure attribution and processing latency. - Added `tests/test_iteration_summary.py` to verify telemetry enhancements and alert correlation. - Documented new metrics in `MONITORING_ALERTING.md` and updated `CHANGELOG.md`. - Fixed `.pip-audit.ignore` formatting to resolve CI dependency audit failures. - Verified system integrity with 32 passing tests. Co-authored-by: xnessom <255288032+xnessom@users.noreply.github.com> --------- Co-authored-by: google-labs-jules[bot] <161369871+google-labs-jules[bot]@users.noreply.github.com> Co-authored-by: xnessom <255288032+xnessom@users.noreply.github.com>
This commit is contained in:
xnessom
co-authored by
xnessom
google-labs-jules[bot] <161369871+google-labs-jules[bot]@users.noreply.github.com>
xnessom
google-labs-jules[bot] <161369871+google-labs-jules[bot]@users.noreply.github.com>
parent
a0bafaba22
commit
5613ac8b73
+2
-2
@@ -1,6 +1,6 @@
|
||||
# Upstream constraint from metaapi-cloud-sdk
|
||||
python - socketio
|
||||
python-socketio
|
||||
# Dependency of metaapi-cloud-sdk
|
||||
python - engineio
|
||||
python-engineio
|
||||
# Known vulnerability in joblib 1.5.3, but required for this environment's scikit-learn version
|
||||
joblib
|
||||
|
||||
@@ -8,6 +8,9 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
|
||||
## [Unreleased]
|
||||
|
||||
### Added
|
||||
- **Structured Iteration Heartbeat:** Implemented a high-fidelity polling cycle heartbeat in `main.py` with failure attribution and processing latency tracking.
|
||||
- **Enhanced Observability Metrics:** Added new Prometheus metrics (`trading_iteration_heartbeat_timestamp`, `trading_iteration_duration_seconds`, `trading_market_stability`) to the `Monitor` module for real-time system health tracking.
|
||||
- **Trace Correlation:** Updated Telegram alerting to automatically include `trace_id` snippets for direct linkage between mobile alerts and system logs.
|
||||
- **Decision Funnel Telemetry:** Implemented a structured signal progression tracking system via `SIGNAL_FUNNEL_COUNTER` to observe decision drop-offs across ensemble, risk, and execution layers.
|
||||
- **Confluence Metrics:** Added `SIGNAL_CONFLUENCE_HISTOGRAM` to track the distribution of weighted confluence scores for institutional signal auditing.
|
||||
- **Security Hardening (Jules02):**
|
||||
|
||||
@@ -366,6 +366,7 @@ The following metrics are exposed by the `Monitor` module:
|
||||
- `trading_slippage_pips`: Difference between expected and actual price (Histogram)
|
||||
- `trading_fill_rate`: Percentage of orders filled at intended price (Gauge)
|
||||
- `trading_orders_rejected_total`: Total number of rejected orders (Counter)
|
||||
- `trading_market_stability`: Real-time market regime confidence/stability score (Gauge)
|
||||
|
||||
### 11.3 System Health
|
||||
- `system_cpu_usage_percent`: System CPU utilization percentage (Gauge)
|
||||
@@ -373,6 +374,8 @@ The following metrics are exposed by the `Monitor` module:
|
||||
- `system_disk_usage_percent`: System disk usage percentage (Gauge)
|
||||
- `trading_system_errors`: Total count of system errors labeled by component (Counter)
|
||||
- `trading_block_duration_seconds`: Duration of trading code blocks labeled by label (Histogram)
|
||||
- `trading_iteration_heartbeat_timestamp`: Unix timestamp of the last iteration heartbeat (Gauge)
|
||||
- `trading_iteration_duration_seconds`: Total duration of a single trading iteration cycle (Histogram)
|
||||
|
||||
### 11.4 Model & Data
|
||||
- `trading_model_confidence`: Latest model prediction confidence (Gauge)
|
||||
|
||||
@@ -214,10 +214,19 @@ def run_live(
|
||||
loop_count = 0
|
||||
last_price = None
|
||||
while True:
|
||||
iteration_start = time.perf_counter()
|
||||
# 0. Generate unique trace ID for this iteration
|
||||
structlog.contextvars.clear_contextvars()
|
||||
structlog.contextvars.bind_contextvars(trace_id=str(uuid.uuid4()))
|
||||
|
||||
# Initialize iteration summary state
|
||||
iteration_status = "processing"
|
||||
final_direction = 0
|
||||
final_confidence = 0.0
|
||||
confluence_score = 0.0
|
||||
market_regime = "unknown"
|
||||
market_stability = 0.0
|
||||
|
||||
# 0. Periodic Audit of Configuration State
|
||||
if loop_count % 100 == 0 and audit_logger:
|
||||
# Dynamic exclusion of all SecretStr/SecretBytes fields
|
||||
@@ -310,13 +319,33 @@ def run_live(
|
||||
monitor.alert_liquidity_crisis(cfg.symbol, spread_pips)
|
||||
|
||||
except MT5DataError as e:
|
||||
iteration_status = "failed_data"
|
||||
log.error("Transient data retrieval error", error=str(e))
|
||||
if monitor:
|
||||
monitor.record_iteration_heartbeat()
|
||||
monitor.record_iteration_duration(time.perf_counter() - iteration_start)
|
||||
log.info(
|
||||
"iteration_summary",
|
||||
status=iteration_status,
|
||||
error=str(e),
|
||||
duration_ms=round((time.perf_counter() - iteration_start) * 1000, 2),
|
||||
)
|
||||
time.sleep(poll_interval)
|
||||
continue
|
||||
except MT5ConnectionError:
|
||||
iteration_status = "failed_connection"
|
||||
log.warning("Connection lost. Attempting reconnection...")
|
||||
if monitor:
|
||||
monitor.alert_broker_connection_lost()
|
||||
monitor.record_iteration_heartbeat()
|
||||
monitor.record_iteration_duration(time.perf_counter() - iteration_start)
|
||||
|
||||
log.info(
|
||||
"iteration_summary",
|
||||
status=iteration_status,
|
||||
duration_ms=round((time.perf_counter() - iteration_start) * 1000, 2),
|
||||
)
|
||||
|
||||
try:
|
||||
connector.connect()
|
||||
log.info("Reconnection successful.")
|
||||
@@ -344,6 +373,8 @@ def run_live(
|
||||
df_features = feature_engineer.compute_features(df_raw)
|
||||
obs = df_features.values[-1] # Full 140+ features
|
||||
regime_info = regime_detector.detect(df_raw)
|
||||
market_regime = regime_info.label.value
|
||||
market_stability = regime_info.confidence
|
||||
|
||||
# Optimization: Extract volatility from already-computed features if available
|
||||
# RegimeDetector already calculates z-score (using rolling std), we can reuse it
|
||||
@@ -364,6 +395,14 @@ def run_live(
|
||||
|
||||
direction = signal_obj.direction
|
||||
confidence = signal_obj.confidence
|
||||
final_direction = direction
|
||||
final_confidence = confidence
|
||||
|
||||
if hasattr(signal_obj, "metadata") and signal_obj.metadata:
|
||||
market_stability = signal_obj.metadata.get(
|
||||
"market_context_stability", market_stability
|
||||
)
|
||||
|
||||
if monitor:
|
||||
monitor.check_confidence_degradation(confidence)
|
||||
# Log model performance if available
|
||||
@@ -540,9 +579,10 @@ def run_live(
|
||||
regime_info=regime_data,
|
||||
execution_data=execution_data,
|
||||
)
|
||||
confluence_score = explanation.get_confluence_score()
|
||||
|
||||
if monitor:
|
||||
monitor.record_confluence(explanation.get_confluence_score())
|
||||
monitor.record_confluence(confluence_score)
|
||||
|
||||
# Log comprehensive decision trace for every non-hold signal
|
||||
if audit_logger:
|
||||
@@ -664,6 +704,26 @@ def run_live(
|
||||
for sym in closed_tickets:
|
||||
risk.open_positions.pop(sym)
|
||||
|
||||
iteration_status = "success"
|
||||
|
||||
# Emit structured iteration summary and record metrics
|
||||
iteration_duration = time.perf_counter() - iteration_start
|
||||
if monitor:
|
||||
monitor.record_iteration_heartbeat()
|
||||
monitor.record_iteration_duration(iteration_duration)
|
||||
monitor.record_market_stability(market_stability)
|
||||
|
||||
log.info(
|
||||
"iteration_summary",
|
||||
status=iteration_status,
|
||||
direction=final_direction,
|
||||
confidence=final_confidence,
|
||||
confluence=confluence_score,
|
||||
regime=market_regime,
|
||||
stability=market_stability,
|
||||
duration_ms=round(iteration_duration * 1000, 2),
|
||||
)
|
||||
|
||||
# Wait for next interval with operator feedback
|
||||
if console:
|
||||
with console.status(
|
||||
@@ -684,9 +744,19 @@ def run_live(
|
||||
)
|
||||
break
|
||||
except MT5ConnectionError as exc:
|
||||
iteration_status = "error_connection"
|
||||
log.error("Critical connection failure: %s. Re-initializing...", exc)
|
||||
if monitor:
|
||||
monitor.alert_broker_connection_lost()
|
||||
monitor.record_iteration_heartbeat()
|
||||
monitor.record_iteration_duration(time.perf_counter() - iteration_start)
|
||||
|
||||
log.info(
|
||||
"iteration_summary",
|
||||
status=iteration_status,
|
||||
error=str(exc),
|
||||
duration_ms=round((time.perf_counter() - iteration_start) * 1000, 2),
|
||||
)
|
||||
time.sleep(5)
|
||||
try:
|
||||
connector.connect()
|
||||
@@ -695,7 +765,18 @@ def run_live(
|
||||
except MT5ConnectionError:
|
||||
log.error("Re-initialization failed during outer loop recovery.")
|
||||
except Exception as exc:
|
||||
iteration_status = "error_unhandled"
|
||||
log.exception("Unhandled error in trading loop: %s", exc)
|
||||
if monitor:
|
||||
monitor.record_iteration_heartbeat()
|
||||
monitor.record_iteration_duration(time.perf_counter() - iteration_start)
|
||||
|
||||
log.info(
|
||||
"iteration_summary",
|
||||
status=iteration_status,
|
||||
error=str(exc),
|
||||
duration_ms=round((time.perf_counter() - iteration_start) * 1000, 2),
|
||||
)
|
||||
time.sleep(poll_interval)
|
||||
|
||||
|
||||
|
||||
@@ -99,6 +99,17 @@ SIGNAL_CONFLUENCE_HISTOGRAM = Histogram(
|
||||
"Weighted confluence score of the trading signal",
|
||||
buckets=(0.1, 0.2, 0.3, 0.4, 0.5, 0.6, 0.7, 0.8, 0.9, 1.0),
|
||||
)
|
||||
MARKET_STABILITY_GAUGE = Gauge(
|
||||
"trading_market_stability", "Real-time market regime confidence/stability score"
|
||||
)
|
||||
|
||||
# 7. Iteration Metrics
|
||||
ITERATION_HEARTBEAT_GAUGE = Gauge(
|
||||
"trading_iteration_heartbeat_timestamp", "Unix timestamp of the last iteration heartbeat"
|
||||
)
|
||||
ITERATION_DURATION_HISTOGRAM = Histogram(
|
||||
"trading_iteration_duration_seconds", "Total duration of a single trading iteration cycle"
|
||||
)
|
||||
|
||||
|
||||
class Monitor:
|
||||
@@ -203,6 +214,13 @@ class Monitor:
|
||||
logger.debug("telegram_not_configured", message=text)
|
||||
return
|
||||
|
||||
# Automatically append trace_id for easier correlation
|
||||
import structlog.contextvars
|
||||
|
||||
trace_id = structlog.contextvars.get_contextvars().get("trace_id")
|
||||
if trace_id:
|
||||
text = f"{text}\n\n[Trace: {trace_id[:8]}]"
|
||||
|
||||
async def _send():
|
||||
try:
|
||||
await self.bot.send_message(chat_id=self.cfg.telegram_chat_id, text=text)
|
||||
@@ -389,6 +407,21 @@ class Monitor:
|
||||
SIGNAL_CONFLUENCE_HISTOGRAM.observe(score)
|
||||
logger.debug("signal_confluence_recorded", score=score)
|
||||
|
||||
def record_market_stability(self, score: float) -> None:
|
||||
"""Record the real-time market stability/regime confidence score."""
|
||||
MARKET_STABILITY_GAUGE.set(score)
|
||||
logger.debug("market_stability_recorded", score=score)
|
||||
|
||||
def record_iteration_heartbeat(self) -> None:
|
||||
"""Record the timestamp of the latest iteration heartbeat."""
|
||||
ITERATION_HEARTBEAT_GAUGE.set_to_current_time()
|
||||
logger.debug("iteration_heartbeat_recorded")
|
||||
|
||||
def record_iteration_duration(self, seconds: float) -> None:
|
||||
"""Record the total duration of the iteration cycle."""
|
||||
ITERATION_DURATION_HISTOGRAM.observe(seconds)
|
||||
logger.debug("iteration_duration_recorded", duration_seconds=seconds)
|
||||
|
||||
def log_model_performance(
|
||||
self, accuracy: float, drift_score: float, calibration_error: float = 0.0
|
||||
) -> None:
|
||||
|
||||
@@ -0,0 +1,60 @@
|
||||
"""
|
||||
Tests for iteration summary logging and telemetry.
|
||||
"""
|
||||
|
||||
import unittest
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
from src.core.monitor import Monitor
|
||||
|
||||
|
||||
class TestIterationSummary(unittest.TestCase):
|
||||
def setUp(self):
|
||||
self.config = MagicMock()
|
||||
self.config.telegram_token.get_secret_value.return_value = "fake_token"
|
||||
self.config.telegram_chat_id = "fake_chat_id"
|
||||
with patch("telegram.Bot"):
|
||||
self.monitor = Monitor(self.config)
|
||||
|
||||
def test_monitor_new_methods(self):
|
||||
"""Test the newly added recording methods in Monitor."""
|
||||
with (
|
||||
patch("src.core.monitor.MARKET_STABILITY_GAUGE.set") as mock_stability,
|
||||
patch(
|
||||
"src.core.monitor.ITERATION_HEARTBEAT_GAUGE.set_to_current_time"
|
||||
) as mock_heartbeat,
|
||||
patch("src.core.monitor.ITERATION_DURATION_HISTOGRAM.observe") as mock_duration,
|
||||
):
|
||||
self.monitor.record_market_stability(0.85)
|
||||
mock_stability.assert_called_once_with(0.85)
|
||||
|
||||
self.monitor.record_iteration_heartbeat()
|
||||
mock_heartbeat.assert_called_once()
|
||||
|
||||
self.monitor.record_iteration_duration(1.23)
|
||||
mock_duration.assert_called_once_with(1.23)
|
||||
|
||||
def test_send_message_with_trace_id(self):
|
||||
"""Test that trace_id is appended to Telegram messages."""
|
||||
self.monitor.bot = MagicMock()
|
||||
|
||||
# Case 1: With trace_id
|
||||
with patch(
|
||||
"structlog.contextvars.get_contextvars", return_value={"trace_id": "test-uuid-12345"}
|
||||
):
|
||||
self.monitor.send_message("Alert message")
|
||||
msg = self.monitor.bot.send_message.call_args[1]["text"]
|
||||
self.assertIn("Alert message", msg)
|
||||
self.assertIn("[Trace: test-uui]", msg)
|
||||
|
||||
self.monitor.bot.send_message.reset_mock()
|
||||
|
||||
# Case 2: Without trace_id
|
||||
with patch("structlog.contextvars.get_contextvars", return_value={}):
|
||||
self.monitor.send_message("Alert message")
|
||||
msg = self.monitor.bot.send_message.call_args[1]["text"]
|
||||
self.assertEqual("Alert message", msg)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
Reference in New Issue
Block a user