mirror of
https://github.com/Sabermrddz/QuantCore-FX.git
synced 2026-07-28 02:57:51 +00:00
saber
564 lines
22 KiB
Python
564 lines
22 KiB
Python
from PyQt5.QtWidgets import (
|
||
QWidget, QVBoxLayout, QHBoxLayout, QLabel, QTableWidget, QTableWidgetItem,
|
||
QPushButton, QComboBox, QCheckBox, QMessageBox,
|
||
QHeaderView
|
||
)
|
||
from PyQt5.QtCore import Qt, QThread, pyqtSignal, QTimer
|
||
from PyQt5.QtGui import QColor, QFont
|
||
from typing import Dict, Optional
|
||
from datetime import datetime, timezone
|
||
import config
|
||
from layer2_technical import TechnicalAnalyzer
|
||
from data_feeder import Mt5DataFeeder
|
||
from currency_strength_matrix import CurrencyStrengthMatrix
|
||
|
||
|
||
class DataStreamerThread(QThread):
|
||
"""Background thread for data source polling."""
|
||
|
||
price_updated = pyqtSignal(dict)
|
||
error_occurred = pyqtSignal(str)
|
||
connected = pyqtSignal(bool)
|
||
|
||
def __init__(self, data_feeder, instruments):
|
||
super().__init__()
|
||
self.feeder = data_feeder
|
||
self.instruments = instruments
|
||
self.running = True
|
||
|
||
def run(self):
|
||
try:
|
||
if not self.feeder.test_connection():
|
||
self.connected.emit(False)
|
||
self.error_occurred.emit("Failed to connect to data source")
|
||
return
|
||
|
||
self.connected.emit(True)
|
||
self.feeder.stream_prices(self.instruments, callback=self._on_price)
|
||
|
||
except Exception as e:
|
||
self.error_occurred.emit(str(e))
|
||
self.connected.emit(False)
|
||
|
||
def _on_price(self, price_data):
|
||
self.price_updated.emit(price_data)
|
||
|
||
def stop(self):
|
||
self.running = False
|
||
if hasattr(self.feeder, 'stop_streaming'):
|
||
self.feeder.stop_streaming()
|
||
|
||
|
||
class Layer2MonitorTab(QWidget):
|
||
"""Layer 2 technical analysis monitoring tab.
|
||
|
||
Task 4.1 — Dynamic Session Visualizations:
|
||
- Active market session indicator (Tokyo / London / New York)
|
||
- High-contrast conditional formatting for ±2σ currency strength cells
|
||
"""
|
||
|
||
def __init__(self, technical_analyzer: TechnicalAnalyzer = None):
|
||
super().__init__()
|
||
|
||
self.tech_analyzer = technical_analyzer or TechnicalAnalyzer()
|
||
self.data_feeder = None
|
||
self.streamer_thread = None
|
||
self.connected = False
|
||
|
||
# Persistent matrix to retain SessionTracker state across refreshes
|
||
self.matrix = CurrencyStrengthMatrix()
|
||
|
||
self._init_ui()
|
||
self._setup_data_source()
|
||
self._refresh_display()
|
||
|
||
def _init_ui(self):
|
||
"""Build UI layout."""
|
||
layout = QVBoxLayout()
|
||
layout.setSpacing(12)
|
||
|
||
# ====== Connection Panel ======
|
||
connection_layout = QHBoxLayout()
|
||
|
||
connection_layout.addWidget(QLabel("Data Source:"))
|
||
self.source_combo = QComboBox()
|
||
self.source_combo.addItems(["MT5 (Live)"])
|
||
self.source_combo.setCurrentIndex(0)
|
||
connection_layout.addWidget(self.source_combo)
|
||
|
||
connection_layout.addWidget(QLabel("TF:"))
|
||
self.tf_combo = QComboBox()
|
||
for tf_key in config.TIMEFRAMES:
|
||
self.tf_combo.addItem(config.TIMEFRAMES[tf_key]["label"], tf_key)
|
||
self.tf_combo.setCurrentText(config.TIMEFRAMES[config.DEFAULT_TIMEFRAME]["label"])
|
||
self.tf_combo.currentIndexChanged.connect(self._on_timeframe_changed)
|
||
connection_layout.addWidget(self.tf_combo)
|
||
|
||
self.connect_btn = QPushButton("Connect")
|
||
self.connect_btn.clicked.connect(self._on_connect_clicked)
|
||
connection_layout.addWidget(self.connect_btn)
|
||
|
||
# Status dot indicator
|
||
self.status_dot = QLabel("●")
|
||
self.status_dot.setStyleSheet("color: #e74c3c; font-size: 18px;")
|
||
connection_layout.addWidget(self.status_dot)
|
||
|
||
self.status_label = QLabel("Disconnected")
|
||
self.status_label.setStyleSheet("color: #e74c3c; font-weight: 600;")
|
||
connection_layout.addWidget(self.status_label)
|
||
|
||
connection_layout.addStretch()
|
||
layout.addLayout(connection_layout)
|
||
|
||
# ====== Active Session Indicator (Task 4.1) ======
|
||
session_layout = QHBoxLayout()
|
||
session_layout.addWidget(QLabel("Active Session:"))
|
||
self.session_label = QLabel("—")
|
||
self.session_label.setStyleSheet(
|
||
"font-weight: 700; font-size: 14px; padding: 4px 14px; "
|
||
"background-color: #ecf0f1; border-radius: 12px;"
|
||
)
|
||
session_layout.addWidget(self.session_label)
|
||
session_layout.addStretch()
|
||
layout.addLayout(session_layout)
|
||
|
||
# ====== Z-Score Table ======
|
||
heading = QLabel("Technical Analysis — All Pairs")
|
||
heading.setProperty("heading", True)
|
||
layout.addWidget(heading)
|
||
|
||
self.tech_table = QTableWidget()
|
||
self.tech_table.setColumnCount(7)
|
||
self.tech_table.setHorizontalHeaderLabels([
|
||
"Pair", "Current Price", "Z-Score", "Volatility", "Mean Price", "Status", "Signal"
|
||
])
|
||
self.tech_table.setRowCount(28)
|
||
self.tech_table.setAlternatingRowColors(True)
|
||
header = self.tech_table.horizontalHeader()
|
||
for c in range(7):
|
||
header.setSectionResizeMode(c, QHeaderView.Stretch)
|
||
|
||
layout.addWidget(self.tech_table)
|
||
|
||
# ====== Alerts Panel ======
|
||
alerts_layout = QHBoxLayout()
|
||
|
||
alerts_layout.addWidget(QLabel("Overbought:"))
|
||
self.overbought_label = QLabel("—")
|
||
self.overbought_label.setStyleSheet("color: #e74c3c; font-weight: 700;")
|
||
alerts_layout.addWidget(self.overbought_label)
|
||
|
||
alerts_layout.addSpacing(20)
|
||
|
||
alerts_layout.addWidget(QLabel("Oversold:"))
|
||
self.oversold_label = QLabel("—")
|
||
self.oversold_label.setStyleSheet("color: #27ae60; font-weight: 700;")
|
||
alerts_layout.addWidget(self.oversold_label)
|
||
|
||
alerts_layout.addStretch()
|
||
layout.addLayout(alerts_layout)
|
||
|
||
# ====== Currency Strength Matrix ======
|
||
matrix_heading = QLabel("Currency Strength Matrix (S.A.T.O.R.I.)")
|
||
matrix_heading.setProperty("heading", True)
|
||
layout.addWidget(matrix_heading)
|
||
|
||
self.matrix_cross_label = QLabel("Matrix Cross: —")
|
||
self.matrix_cross_label.setStyleSheet("font-weight: 700; font-size: 14px;")
|
||
layout.addWidget(self.matrix_cross_label)
|
||
|
||
self.divergence_label = QLabel("Divergence Gap: 0.0")
|
||
self.divergence_label.setStyleSheet("color: #7f8c8d; font-size: 12px;")
|
||
layout.addWidget(self.divergence_label)
|
||
|
||
alert_row = QHBoxLayout()
|
||
self.strong_alert = QLabel("")
|
||
self.strong_alert.setStyleSheet("color: #27ae60; font-weight: 600;")
|
||
alert_row.addWidget(self.strong_alert)
|
||
self.weak_alert = QLabel("")
|
||
self.weak_alert.setStyleSheet("color: #e74c3c; font-weight: 600;")
|
||
alert_row.addWidget(self.weak_alert)
|
||
alert_row.addStretch()
|
||
layout.addLayout(alert_row)
|
||
|
||
self.matrix_table = QTableWidget()
|
||
self.matrix_table.setColumnCount(5)
|
||
self.matrix_table.setHorizontalHeaderLabels([
|
||
"Rank", "Currency", "Strength Z", "Direction", "Session SRV"
|
||
])
|
||
self.matrix_table.setRowCount(8)
|
||
self.matrix_table.setMaximumHeight(220)
|
||
m_header = self.matrix_table.horizontalHeader()
|
||
for c in range(5):
|
||
m_header.setSectionResizeMode(c, QHeaderView.Stretch)
|
||
layout.addWidget(self.matrix_table)
|
||
|
||
# ====== Controls ======
|
||
button_layout = QHBoxLayout()
|
||
|
||
self.auto_refresh_check = QCheckBox("Auto-refresh (every 1s)")
|
||
self.auto_refresh_check.setChecked(True)
|
||
button_layout.addWidget(self.auto_refresh_check)
|
||
|
||
refresh_btn = QPushButton("Refresh Now")
|
||
refresh_btn.setObjectName("secondary")
|
||
refresh_btn.clicked.connect(self._refresh_display)
|
||
button_layout.addWidget(refresh_btn)
|
||
|
||
button_layout.addStretch()
|
||
layout.addLayout(button_layout)
|
||
layout.addStretch()
|
||
|
||
self.setLayout(layout)
|
||
|
||
self.refresh_timer = QTimer()
|
||
self.refresh_timer.timeout.connect(self._refresh_display)
|
||
|
||
self._debounce_timer = QTimer()
|
||
self._debounce_timer.setSingleShot(True)
|
||
self._debounce_timer.timeout.connect(self._refresh_display)
|
||
|
||
def _setup_data_source(self):
|
||
"""Initialize data source."""
|
||
self.data_feeder = Mt5DataFeeder()
|
||
|
||
def _on_connect_clicked(self):
|
||
"""Handle connect button click."""
|
||
if self.connected:
|
||
self._disconnect()
|
||
else:
|
||
self._connect()
|
||
|
||
def _on_timeframe_changed(self, idx: int):
|
||
tf_key = self.tf_combo.itemData(idx)
|
||
if tf_key:
|
||
self.tech_analyzer.set_timeframe(tf_key)
|
||
if self.connected and hasattr(self.data_feeder, 'USD_PAIRS'):
|
||
self._seed_historical_bars()
|
||
|
||
def _seed_historical_bars(self):
|
||
"""Seed the analyzer with bar data at the selected timeframe.
|
||
Fetches only the 7 USD pairs and derives all cross rates.
|
||
"""
|
||
if not hasattr(self.data_feeder, 'USD_PAIRS'):
|
||
return
|
||
|
||
interval_map = {
|
||
"M5": "5min", "M15": "15min",
|
||
"H1": "1h", "H4": "4h",
|
||
}
|
||
tf_key = self.tech_analyzer.current_timeframe
|
||
interval = interval_map.get(tf_key, "15min")
|
||
outputsize = "full" if tf_key in ("M5", "M15") else "full"
|
||
|
||
usd_pairs = self.data_feeder.USD_PAIRS
|
||
raw_ohlc: dict[str, list[dict]] = {}
|
||
for pair in usd_pairs:
|
||
base, quote = pair.split("_")
|
||
candles = self.data_feeder.get_historical_candles(
|
||
from_currency=base, to_currency=quote,
|
||
interval=interval, outputsize=outputsize
|
||
)
|
||
if candles:
|
||
raw_ohlc[pair] = candles
|
||
|
||
if not raw_ohlc:
|
||
return
|
||
|
||
n_bars = min(len(c) for c in raw_ohlc.values())
|
||
if n_bars < 2:
|
||
return
|
||
|
||
currencies = config.CURRENCIES
|
||
derived_ohlc: dict[str, list[dict]] = {}
|
||
for base in currencies:
|
||
for quote in currencies:
|
||
if base == quote:
|
||
continue
|
||
derived_ohlc[f"{base}_{quote}"] = []
|
||
|
||
for i in range(n_bars):
|
||
usd_rates: dict[str, float] = {"USD": 1.0}
|
||
for pair in usd_pairs:
|
||
base, quote = pair.split("_")
|
||
c = raw_ohlc[pair][i]
|
||
mid = c["close"]
|
||
if base == "USD":
|
||
usd_rates[quote] = 1.0 / mid if mid else None
|
||
else:
|
||
usd_rates[base] = mid
|
||
for base in currencies:
|
||
bv = usd_rates.get(base)
|
||
if bv is None:
|
||
continue
|
||
for quote in currencies:
|
||
if base == quote:
|
||
continue
|
||
qv = usd_rates.get(quote)
|
||
if qv is not None:
|
||
rate = bv / qv
|
||
# Estimate OHLC for the cross pair
|
||
derived_ohlc[f"{base}_{quote}"].append({
|
||
"close": rate,
|
||
"high": rate * 1.0003,
|
||
"low": rate * 0.9997,
|
||
})
|
||
|
||
self.tech_analyzer.seed_ohlc(derived_ohlc)
|
||
|
||
def _connect(self):
|
||
"""Connect to data source."""
|
||
try:
|
||
if not self.data_feeder.test_connection():
|
||
reason = getattr(self.data_feeder, 'last_error', 'Unknown error')
|
||
QMessageBox.warning(self, "Connection Error", f"Failed to connect to data source:\n{reason}")
|
||
return
|
||
|
||
instruments = self.data_feeder.get_all_major_pairs()
|
||
self.streamer_thread = DataStreamerThread(self.data_feeder, instruments)
|
||
self.streamer_thread.price_updated.connect(self._on_price_received)
|
||
self.streamer_thread.error_occurred.connect(self._on_streamer_error)
|
||
self.streamer_thread.connected.connect(self._on_connected)
|
||
self.streamer_thread.start()
|
||
|
||
if self.auto_refresh_check.isChecked():
|
||
self.refresh_timer.start(3000)
|
||
|
||
self.connected = True
|
||
self.connect_btn.setText("Disconnect")
|
||
self.connect_btn.setObjectName("danger")
|
||
self.connect_btn.style().unpolish(self.connect_btn)
|
||
self.connect_btn.style().polish(self.connect_btn)
|
||
self.status_dot.setStyleSheet("color: #27ae60; font-size: 18px;")
|
||
self.status_label.setText("Connected")
|
||
self.status_label.setStyleSheet("color: #27ae60; font-weight: 600;")
|
||
self._refresh_display()
|
||
|
||
QTimer.singleShot(0, self._seed_historical_bars)
|
||
|
||
except Exception as e:
|
||
QMessageBox.critical(self, "Error", f"Connection failed: {e}")
|
||
|
||
def _disconnect(self):
|
||
"""Disconnect from data source."""
|
||
if self.streamer_thread:
|
||
self.streamer_thread.stop()
|
||
self.streamer_thread.quit()
|
||
self.streamer_thread.wait()
|
||
|
||
self.refresh_timer.stop()
|
||
|
||
self.connected = False
|
||
self.connect_btn.setText("Connect")
|
||
self.connect_btn.setObjectName("")
|
||
self.connect_btn.style().unpolish(self.connect_btn)
|
||
self.connect_btn.style().polish(self.connect_btn)
|
||
self.status_dot.setStyleSheet("color: #e74c3c; font-size: 18px;")
|
||
self.status_label.setText("Disconnected")
|
||
self.status_label.setStyleSheet("color: #e74c3c; font-weight: 600;")
|
||
|
||
def _on_price_received(self, price_data):
|
||
"""Handle price update — just add data, debounce display refresh."""
|
||
pair = price_data.get('pair')
|
||
mid_price = price_data.get('mid')
|
||
|
||
if pair and mid_price:
|
||
self.tech_analyzer.add_price_data(pair, mid_price)
|
||
if not self._debounce_timer.isActive():
|
||
self._debounce_timer.start(2000)
|
||
|
||
def _on_connected(self, is_connected):
|
||
"""Handle connection status change."""
|
||
if is_connected:
|
||
self.status_label.setText("Connected")
|
||
self.status_label.setStyleSheet("color: #27ae60; font-weight: bold;")
|
||
else:
|
||
self.status_label.setText("Disconnected")
|
||
self.status_label.setStyleSheet("color: #e74c3c; font-weight: bold;")
|
||
|
||
def _on_streamer_error(self, error_msg):
|
||
"""Handle streamer error."""
|
||
print(f"[Layer2] Streamer error: {error_msg}")
|
||
|
||
def _refresh_display(self):
|
||
"""Refresh the technical analysis display."""
|
||
try:
|
||
z_scores = self.tech_analyzer.get_all_z_scores()
|
||
overbought = self.tech_analyzer.get_overbought_pairs()
|
||
oversold = self.tech_analyzer.get_oversold_pairs()
|
||
|
||
for row, (pair, z_score) in enumerate(sorted(z_scores.items())):
|
||
if row >= self.tech_table.rowCount():
|
||
break
|
||
|
||
status = self.tech_analyzer.get_status_for_pair(pair)
|
||
|
||
pair_item = QTableWidgetItem(pair)
|
||
pair_item.setFlags(pair_item.flags() & ~Qt.ItemIsEditable)
|
||
self.tech_table.setItem(row, 0, pair_item)
|
||
|
||
last_price = self.tech_analyzer.get_last_price(pair)
|
||
price_text = f"{last_price:.4f}" if last_price else "—"
|
||
price_item = QTableWidgetItem(price_text)
|
||
price_item.setFlags(price_item.flags() & ~Qt.ItemIsEditable)
|
||
self.tech_table.setItem(row, 1, price_item)
|
||
|
||
z_item = QTableWidgetItem(f"{z_score:.2f}")
|
||
z_item.setFlags(z_item.flags() & ~Qt.ItemIsEditable)
|
||
z_item.setTextAlignment(Qt.AlignCenter)
|
||
|
||
if abs(z_score) >= config.SCALP_Z_SCORE_THRESHOLD:
|
||
z_item.setBackground(QColor("#ffebee"))
|
||
z_item.setForeground(QColor("#c62828"))
|
||
|
||
self.tech_table.setItem(row, 2, z_item)
|
||
|
||
vol_item = QTableWidgetItem(f"{status['volatility']:.4f}")
|
||
vol_item.setFlags(vol_item.flags() & ~Qt.ItemIsEditable)
|
||
self.tech_table.setItem(row, 3, vol_item)
|
||
|
||
mean_item = QTableWidgetItem(f"{status['mean_price']:.4f}")
|
||
mean_item.setFlags(mean_item.flags() & ~Qt.ItemIsEditable)
|
||
self.tech_table.setItem(row, 4, mean_item)
|
||
|
||
status_item = QTableWidgetItem(status['status'])
|
||
status_item.setFlags(status_item.flags() & ~Qt.ItemIsEditable)
|
||
|
||
if "OVERBOUGHT" in status['status']:
|
||
status_item.setBackground(QColor("#ffebee"))
|
||
elif "OVERSOLD" in status['status']:
|
||
status_item.setBackground(QColor("#e8f5e9"))
|
||
|
||
self.tech_table.setItem(row, 5, status_item)
|
||
|
||
if status['is_extreme']:
|
||
signal = "EXTREME"
|
||
signal_item = QTableWidgetItem(signal)
|
||
signal_item.setBackground(QColor("#fff3e0"))
|
||
else:
|
||
signal = "Normal"
|
||
signal_item = QTableWidgetItem(signal)
|
||
|
||
signal_item.setFlags(signal_item.flags() & ~Qt.ItemIsEditable)
|
||
self.tech_table.setItem(row, 6, signal_item)
|
||
|
||
overbought_text = ", ".join(overbought) if overbought else "None"
|
||
oversold_text = ", ".join(oversold) if oversold else "None"
|
||
|
||
self.overbought_label.setText(overbought_text)
|
||
self.oversold_label.setText(oversold_text)
|
||
|
||
# ====== Currency Strength Matrix (persistent instance) ======
|
||
current_prices = {}
|
||
for pair in z_scores:
|
||
lp = self.tech_analyzer.get_last_price(pair)
|
||
if lp is not None:
|
||
current_prices[pair] = lp
|
||
self.matrix.update(z_scores, current_prices=current_prices)
|
||
report = self.matrix.get_report()
|
||
|
||
matrix_cross = report["matrix_cross"]
|
||
gap = report["divergence_gap"]
|
||
self.matrix_cross_label.setText(
|
||
f"Matrix Cross: {matrix_cross or '—'} | Spread: {gap:.2f}σ"
|
||
)
|
||
|
||
if report["has_divergence"]:
|
||
self.matrix_cross_label.setStyleSheet(
|
||
"font-weight: bold; font-size: 13px; color: #e74c3c;"
|
||
)
|
||
self.divergence_label.setText(
|
||
"DIVERGENCE DETECTED — extreme strength vs extreme weakness"
|
||
)
|
||
self.divergence_label.setStyleSheet("color: #e74c3c; font-weight: bold;")
|
||
else:
|
||
self.matrix_cross_label.setStyleSheet(
|
||
"font-weight: bold; font-size: 13px; color: #2c3e50;"
|
||
)
|
||
self.divergence_label.setText("No extreme divergence")
|
||
self.divergence_label.setStyleSheet("color: #7f8c8d;")
|
||
|
||
ob_currencies = report["overbought"]
|
||
os_currencies = report["oversold"]
|
||
self.strong_alert.setText(
|
||
f"Overbought Currencies: {', '.join(ob_currencies) if ob_currencies else 'None'}"
|
||
)
|
||
self.weak_alert.setText(
|
||
f"Oversold Currencies: {', '.join(os_currencies) if os_currencies else 'None'}"
|
||
)
|
||
|
||
# ====== Active Session Indicator (Task 4.1) ======
|
||
active_session = report.get("active_session", "—")
|
||
session_colors = {
|
||
"Tokyo": "#8e44ad",
|
||
"London": "#2980b9",
|
||
"New York": "#e67e22",
|
||
"Off-Hours": "#7f8c8d",
|
||
}
|
||
session_color = session_colors.get(active_session, "#7f8c8d")
|
||
self.session_label.setText(active_session)
|
||
self.session_label.setStyleSheet(
|
||
f"font-weight: bold; font-size: 14px; padding: 2px 8px; "
|
||
f"color: white; background-color: {session_color}; "
|
||
f"border-radius: 4px;"
|
||
)
|
||
|
||
# ====== Ranked Currency Table with High-Contrast σ (Task 4.1) ======
|
||
ranked = report["ranked"]
|
||
for row, entry in enumerate(ranked):
|
||
ccy = entry[0]
|
||
z_val = entry[1]
|
||
direction = entry[2]
|
||
srv = entry[3] if len(entry) > 3 else 0.0
|
||
|
||
rank_item = QTableWidgetItem(str(row + 1))
|
||
rank_item.setFlags(rank_item.flags() & ~Qt.ItemIsEditable)
|
||
rank_item.setTextAlignment(Qt.AlignCenter)
|
||
self.matrix_table.setItem(row, 0, rank_item)
|
||
|
||
ccy_item = QTableWidgetItem(ccy)
|
||
ccy_item.setFlags(ccy_item.flags() & ~Qt.ItemIsEditable)
|
||
self.matrix_table.setItem(row, 1, ccy_item)
|
||
|
||
z_item = QTableWidgetItem(f"{z_val:.2f}")
|
||
z_item.setFlags(z_item.flags() & ~Qt.ItemIsEditable)
|
||
z_item.setTextAlignment(Qt.AlignCenter)
|
||
|
||
# High-contrast σ formatting (Task 4.1)
|
||
threshold = config.SCALP_Z_SCORE_THRESHOLD
|
||
if z_val >= threshold:
|
||
z_item.setBackground(QColor("#c62828"))
|
||
z_item.setForeground(QColor("white"))
|
||
elif z_val <= -threshold:
|
||
z_item.setBackground(QColor("#2e7d32"))
|
||
z_item.setForeground(QColor("white"))
|
||
|
||
self.matrix_table.setItem(row, 2, z_item)
|
||
|
||
dir_item = QTableWidgetItem(direction)
|
||
dir_item.setFlags(dir_item.flags() & ~Qt.ItemIsEditable)
|
||
if direction == "OVERBOUGHT":
|
||
dir_item.setBackground(QColor("#ffebee"))
|
||
dir_item.setForeground(QColor("#c62828"))
|
||
elif direction == "OVERSOLD":
|
||
dir_item.setBackground(QColor("#e8f5e9"))
|
||
dir_item.setForeground(QColor("#2e7d32"))
|
||
self.matrix_table.setItem(row, 3, dir_item)
|
||
|
||
# Session Relative Velocity column
|
||
srv_sign = "+" if srv >= 0 else ""
|
||
srv_item = QTableWidgetItem(f"{srv_sign}{srv:.4f}%")
|
||
srv_item.setFlags(srv_item.flags() & ~Qt.ItemIsEditable)
|
||
srv_item.setTextAlignment(Qt.AlignCenter)
|
||
if abs(srv) > 0.5:
|
||
srv_item.setBackground(QColor("#fff3e0"))
|
||
self.matrix_table.setItem(row, 4, srv_item)
|
||
|
||
except Exception as e:
|
||
print(f"[Layer2] Display error: {e}")
|
||
|
||
def closeEvent(self, event):
|
||
"""Clean up on close."""
|
||
self._disconnect()
|
||
event.accept()
|