276 lines
15 KiB
Python
276 lines
15 KiB
Python
"""
|
|
Models & Training Progress Panel - Clean Implementation
|
|
Displays real-time model status, training metrics, and performance data
|
|
"""
|
|
|
|
import logging
|
|
from typing import Dict, Any, List, Optional
|
|
from datetime import datetime
|
|
from dash import html, dcc
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class ModelsTrainingPanel:
|
|
"""Clean implementation of the Models & Training Progress panel.
|
|
- Uses orchestrator.get_model_statistics_summary() and model_states
|
|
- No synthetic or placeholder data; shows empty when unavailable
|
|
- Avoids duplicate class definitions and invalid imports
|
|
"""
|
|
|
|
def __init__(self, orchestrator=None):
|
|
self.orchestrator = orchestrator
|
|
|
|
def create_panel(self) -> html.Div:
|
|
try:
|
|
data = self._gather_data()
|
|
|
|
content: List[html.Div] = []
|
|
content.append(self._create_header())
|
|
|
|
content.append(self._create_models_section(data.get("models", {})))
|
|
|
|
if data.get("training_status"):
|
|
content.append(self._create_training_status_section(data["training_status"]))
|
|
|
|
if data.get("system_metrics"):
|
|
content.append(self._create_system_metrics_section(data["system_metrics"]))
|
|
|
|
return html.Div(content, id="training-metrics")
|
|
except Exception as e:
|
|
logger.error(f"Error creating models training panel: {e}")
|
|
return html.Div([
|
|
html.P(f"Error loading training panel: {str(e)}", className="text-danger small")
|
|
], id="training-metrics")
|
|
|
|
def _gather_data(self) -> Dict[str, Any]:
|
|
result: Dict[str, Any] = {"models": {}, "training_status": {}, "system_metrics": {}}
|
|
if not self.orchestrator:
|
|
return result
|
|
try:
|
|
# Model statistics summary (serializable)
|
|
stats_summary: Dict[str, Dict[str, Any]] = {}
|
|
if hasattr(self.orchestrator, "get_model_statistics_summary"):
|
|
stats = self.orchestrator.get_model_statistics_summary()
|
|
if isinstance(stats, dict):
|
|
stats_summary = stats
|
|
|
|
# Model states (for best_loss and checkpoint flags)
|
|
model_states: Dict[str, Dict[str, Any]] = getattr(self.orchestrator, "model_states", {}) or {}
|
|
|
|
# Build models block from stats_summary
|
|
for model_name, s in stats_summary.items():
|
|
model_info: Dict[str, Any] = {}
|
|
try:
|
|
# Status: active if we have any inference info
|
|
total_inf = int(s.get("total_inferences", 0) or 0)
|
|
status = "active" if (total_inf > 0 or s.get("last_inference_time")) else "registered"
|
|
|
|
# Last prediction
|
|
last_pred_action = s.get("last_prediction")
|
|
last_confidence = s.get("last_confidence")
|
|
last_inference_time = s.get("last_inference_time")
|
|
|
|
# Loss metrics
|
|
loss_metrics = {
|
|
"current_loss": s.get("current_loss"),
|
|
"best_loss": s.get("best_loss"),
|
|
}
|
|
|
|
# Timing metrics
|
|
timing_metrics = {
|
|
"inferences_per_second": s.get("inference_rate_per_second"),
|
|
"last_inference": last_inference_time,
|
|
"last_training": s.get("last_training_time"),
|
|
}
|
|
|
|
# Checkpoint flags from orchestrator.model_states if present
|
|
ckpt_loaded = False
|
|
ckpt_failed = False
|
|
if model_states and model_name in model_states:
|
|
ckpt_loaded = bool(model_states[model_name].get("checkpoint_loaded", False))
|
|
ckpt_failed = bool(model_states[model_name].get("checkpoint_failed", False))
|
|
# If model started fresh, mark as stored after first successful autosave/best_loss update
|
|
if not ckpt_loaded:
|
|
cur = model_states[model_name].get("current_loss")
|
|
best = model_states[model_name].get("best_loss")
|
|
if isinstance(cur, (int, float)) and isinstance(best, (int, float)) and best is not None and cur is not None and cur <= best:
|
|
ckpt_loaded = True
|
|
checkpoint_filename = model_states[model_name].get("checkpoint_filename")
|
|
|
|
model_info = {
|
|
"name": model_name,
|
|
"status": status,
|
|
"parameters": 0, # unknown; do not synthesize
|
|
"last_prediction": {
|
|
"action": last_pred_action if last_pred_action is not None else "",
|
|
"confidence": last_confidence if last_confidence is not None else None,
|
|
"timestamp": last_inference_time if last_inference_time else "",
|
|
},
|
|
"training_enabled": True,
|
|
"inference_enabled": True,
|
|
"routing_enabled": True,
|
|
"checkpoint_loaded": ckpt_loaded,
|
|
"checkpoint_failed": ckpt_failed,
|
|
"checkpoint_filename": checkpoint_filename if model_states and model_name in model_states else None,
|
|
"loss_metrics": loss_metrics,
|
|
"timing_metrics": timing_metrics,
|
|
"signal_stats": {},
|
|
}
|
|
except Exception as e:
|
|
logger.debug(f"Error building model info for {model_name}: {e}")
|
|
model_info = {"name": model_name, "status": "error", "error": str(e)}
|
|
|
|
result["models"][model_name] = model_info
|
|
|
|
# Enhanced training system status
|
|
training_status: Dict[str, Any] = {}
|
|
try:
|
|
if hasattr(self.orchestrator, "get_enhanced_training_stats"):
|
|
training_status = self.orchestrator.get_enhanced_training_stats() or {}
|
|
except Exception as e:
|
|
logger.debug(f"Enhanced training stats unavailable: {e}")
|
|
result["training_status"] = training_status
|
|
|
|
# System metrics (decision fusion, cob integration)
|
|
system_metrics = {
|
|
"decision_fusion_active": bool(getattr(self.orchestrator, "decision_fusion_enabled", False)),
|
|
"cob_integration_active": bool(getattr(self.orchestrator, "cob_integration", None) is not None),
|
|
"symbols_tracking": len(getattr(getattr(self.orchestrator, "cob_integration", None), "symbols", []) or []),
|
|
}
|
|
result["system_metrics"] = system_metrics
|
|
|
|
return result
|
|
except Exception as e:
|
|
logger.error(f"Error gathering panel data: {e}")
|
|
return result
|
|
|
|
def _create_header(self) -> html.Div:
|
|
return html.Div([
|
|
html.H6([html.I(className="fas fa-brain me-2 text-primary"), "Models & Training Progress"], className="mb-2"),
|
|
html.Button([html.I(className="fas fa-sync-alt me-1"), "Refresh"], id="refresh-training-metrics-btn", className="btn btn-sm btn-outline-primary mb-2")
|
|
], className="d-flex justify-content-between align-items-start")
|
|
|
|
def _create_models_section(self, models: Dict[str, Any]) -> html.Div:
|
|
cards = [self._create_model_card(name, info) for name, info in models.items()]
|
|
return html.Div([
|
|
html.H6([html.I(className="fas fa-microchip me-2 text-success"), f"Loaded Models ({len(models)})"], className="mb-2"),
|
|
html.Div(cards)
|
|
])
|
|
|
|
def _create_model_card(self, model_name: str, model_data: Dict[str, Any]) -> html.Div:
|
|
status = model_data.get("status", "unknown")
|
|
if status == "active":
|
|
status_class = "text-success"; status_icon = "fas fa-check-circle"; status_text = "ACTIVE"
|
|
elif status == "registered":
|
|
status_class = "text-warning"; status_icon = "fas fa-circle"; status_text = "REGISTERED"
|
|
elif status == "inactive":
|
|
status_class = "text-muted"; status_icon = "fas fa-pause-circle"; status_text = "INACTIVE"
|
|
else:
|
|
status_class = "text-danger"; status_icon = "fas fa-exclamation-circle"; status_text = "UNKNOWN"
|
|
|
|
last_pred = model_data.get("last_prediction", {})
|
|
pred_action = last_pred.get("action") or "NONE"
|
|
pred_confidence = last_pred.get("confidence")
|
|
pred_time = last_pred.get("timestamp") or "N/A"
|
|
|
|
loss_metrics = model_data.get("loss_metrics", {})
|
|
current_loss = loss_metrics.get("current_loss")
|
|
best_loss = loss_metrics.get("best_loss")
|
|
loss_class = (
|
|
"text-success" if (isinstance(current_loss, (int, float)) and current_loss < 0.1)
|
|
else "text-warning" if (isinstance(current_loss, (int, float)) and current_loss < 0.5)
|
|
else "text-danger"
|
|
) if current_loss is not None else "text-muted"
|
|
|
|
timing = model_data.get("timing_metrics", {})
|
|
rate = timing.get("inferences_per_second")
|
|
|
|
# Tooltip title showing checkpoint info (if any)
|
|
title_parts: List[str] = [f"{status_text}"]
|
|
ckpt_name = model_data.get('checkpoint_filename')
|
|
if ckpt_name:
|
|
title_parts.append(f"CKPT: {ckpt_name}")
|
|
header_title = " | ".join(title_parts)
|
|
|
|
return html.Div([
|
|
html.Div([
|
|
html.Div([
|
|
html.I(className=f"{status_icon} me-2 {status_class}"),
|
|
html.Strong(f"{model_name.upper()}", className=status_class, title=header_title),
|
|
html.Span(f" - {status_text}", className=f"{status_class} small ms-1"),
|
|
html.Span(" [CKPT]" if model_data.get("checkpoint_loaded") else (" [FAILED]" if model_data.get("checkpoint_failed") else " [FRESH]"), className=f"small {'text-success' if model_data.get('checkpoint_loaded') else 'text-danger' if model_data.get('checkpoint_failed') else 'text-warning'} ms-1")
|
|
], style={"flex": "1"}),
|
|
html.Div([
|
|
html.Div([
|
|
html.Label("Inf", className="text-muted small me-1", style={"fontSize": "10px"}),
|
|
dcc.Checklist(id={'type': 'model-toggle', 'model': model_name, 'toggle_type': 'inference'}, options=[{"label": "", "value": True}], value=[True] if model_data.get('inference_enabled', True) else [], className="form-check-input me-2", style={"transform": "scale(0.7)"})
|
|
], className="d-flex align-items-center me-2"),
|
|
html.Div([
|
|
html.Label("Trn", className="text-muted small me-1", style={"fontSize": "10px"}),
|
|
dcc.Checklist(id={'type': 'model-toggle', 'model': model_name, 'toggle_type': 'training'}, options=[{"label": "", "value": True}], value=[True] if model_data.get('training_enabled', True) else [], className="form-check-input", style={"transform": "scale(0.7)"})
|
|
], className="d-flex align-items-center me-2"),
|
|
html.Div([
|
|
html.Label("Route", className="text-muted small me-1", style={"fontSize": "10px"}),
|
|
dcc.Checklist(id={'type': 'model-toggle', 'model': model_name, 'toggle_type': 'routing'}, options=[{"label": "", "value": True}], value=[True] if model_data.get('routing_enabled', True) else [], className="form-check-input", style={"transform": "scale(0.7)"})
|
|
], className="d-flex align-items-center"),
|
|
], className="d-flex")
|
|
], className="d-flex align-items-center mb-2"),
|
|
|
|
html.Div([
|
|
html.Div([
|
|
html.Span("Last: ", className="text-muted small"),
|
|
html.Span(f"{pred_action}", className=f"small fw-bold {'text-success' if pred_action == 'BUY' else 'text-danger' if pred_action == 'SELL' else 'text-warning'}"),
|
|
html.Span(f" ({pred_confidence:.1f}%)" if isinstance(pred_confidence, (int, float)) else "", className="text-muted small"),
|
|
html.Span(f" @ {pred_time}", className="text-muted small")
|
|
], className="mb-1"),
|
|
html.Div([
|
|
html.Span("Loss: ", className="text-muted small"),
|
|
html.Span(f"{current_loss:.4f}" if isinstance(current_loss, (int, float)) else "", className=f"small fw-bold {loss_class}"),
|
|
*([html.Span(" | Best: ", className="text-muted small"), html.Span(f"{best_loss:.4f}", className="text-success small")] if isinstance(best_loss, (int, float)) else [])
|
|
], className="mb-1"),
|
|
html.Div([
|
|
html.Span("Rate: ", className="text-muted small"),
|
|
html.Span(f"{rate:.2f}/s" if isinstance(rate, (int, float)) else "", className="text-info small")
|
|
], className="mb-1"),
|
|
html.Div([
|
|
html.Span("Last Inf: ", className="text-muted small"),
|
|
html.Span(f"{timing.get('last_inference', 'N/A')}", className="text-info small"),
|
|
html.Span(" | Train: ", className="text-muted small"),
|
|
html.Span(f"{timing.get('last_training', 'N/A')}", className="text-warning small")
|
|
], className="mb-1"),
|
|
])
|
|
], className="border rounded p-2 mb-2")
|
|
|
|
def _create_training_status_section(self, status: Dict[str, Any]) -> html.Div:
|
|
if not status or status.get("status") == "error":
|
|
return html.Div([
|
|
html.Hr(),
|
|
html.H6([html.I(className="fas fa-exclamation-triangle me-2 text-danger"), "Training Status Error"], className="mb-2"),
|
|
html.P(f"Error: {status.get('error', 'Unknown')}", className="text-danger small")
|
|
])
|
|
is_enabled = status.get("training_enabled", False)
|
|
return html.Div([
|
|
html.Hr(),
|
|
html.H6([html.I(className="fas fa-brain me-2 text-secondary"), "Training Status"], className="mb-2"),
|
|
html.Div([
|
|
html.Span("Enabled: ", className="text-muted small"),
|
|
html.Span("ON" if is_enabled else "OFF", className=f"small fw-bold {'text-success' if is_enabled else 'text-warning'}"),
|
|
], className="mb-1"),
|
|
])
|
|
|
|
def _create_system_metrics_section(self, metrics: Dict[str, Any]) -> html.Div:
|
|
return html.Div([
|
|
html.Hr(),
|
|
html.H6([html.I(className="fas fa-chart-line me-2 text-primary"), "System Performance"], className="mb-2"),
|
|
html.Div([
|
|
html.Span("Decision Fusion: ", className="text-muted small"),
|
|
html.Span("ON" if metrics.get("decision_fusion_active") else "OFF", className=f"small {'text-success' if metrics.get('decision_fusion_active') else 'text-muted'}"),
|
|
html.Span(" | COB: ", className="text-muted small"),
|
|
html.Span("ON" if metrics.get("cob_integration_active") else "OFF", className=f"small {'text-success' if metrics.get('cob_integration_active') else 'text-muted'}"),
|
|
], className="mb-1"),
|
|
html.Div([
|
|
html.Span("Tracking: ", className="text-muted small"),
|
|
html.Span(f"{metrics.get('symbols_tracking', 0)} symbols", className="text-info small"),
|
|
], className="mb-0"),
|
|
]) |