from __future__ import annotations from PyQt6.QtWidgets import ( QWidget, QVBoxLayout, QHBoxLayout, QLabel, QPushButton, QTableWidget, QTableWidgetItem, QDoubleSpinBox, QComboBox, QGroupBox, QFormLayout, QCheckBox, QProgressBar, QMenu ) from PyQt6.QtCore import Qt, QThread, pyqtSignal from PyQt6.QtGui import QColor from utils.formatters import fmt_currency, fmt_percent, fmt_large_number, pnl_color SECTORS = ["All", "Technology", "Healthcare", "Financials", "Consumer Cyclical", "Communication Services", "Industrials", "Consumer Defensive", "Energy", "Basic Materials", "Real Estate", "Utilities", "Crypto"] class ScreenerWorker(QThread): data_ready = pyqtSignal(list) progress = pyqtSignal(int) error = pyqtSignal(str) def __init__(self, filters: dict): super().__init__() self._filters = filters def run(self): try: from core.market_data import get_screener_data, SCREENER_UNIVERSE raw = get_screener_data(SCREENER_UNIVERSE) self.progress.emit(80) filtered = self._apply_filters(raw) self.data_ready.emit(filtered) except Exception as e: self.error.emit(str(e)) def _apply_filters(self, data: list[dict]) -> list[dict]: f = self._filters result = [] for row in data: if f.get("sector") and f["sector"] != "All": if row.get("sector", "") != f["sector"]: continue if row.get("pe_ratio") is not None: pe = row["pe_ratio"] if f.get("pe_min") and pe < f["pe_min"]: continue if f.get("pe_max") and pe > f["pe_max"]: continue if row.get("market_cap") is not None: mc = row["market_cap"] if f.get("mc_min") and mc < f["mc_min"] * 1e9: continue if f.get("mc_max") and mc > f["mc_max"] * 1e9: continue if row.get("rsi") is not None: rsi = row["rsi"] if f.get("rsi_min") and rsi < f["rsi_min"]: continue if f.get("rsi_max") and rsi > f["rsi_max"]: continue if row.get("pct_from_52h") is not None: pct = row["pct_from_52h"] if f.get("from_52h_min") and pct < f["from_52h_min"]: continue if f.get("from_52h_max") and pct > f["from_52h_max"]: continue result.append(row) return result class ScreenerWidget(QWidget): symbol_selected = pyqtSignal(str) add_to_watchlist = pyqtSignal(str) def __init__(self, parent=None): super().__init__(parent) self._worker: ScreenerWorker | None = None self._results: list[dict] = [] self._setup_ui() def _setup_ui(self): root = QVBoxLayout(self) root.setContentsMargins(12, 12, 12, 12) root.setSpacing(10) header = QLabel("Stock Screener") header.setObjectName("title") root.addWidget(header) # Filters filters_box = QGroupBox("Filters") filters_layout = QHBoxLayout(filters_box) # Sector sector_form = QFormLayout() self._sector_combo = QComboBox() self._sector_combo.addItems(SECTORS) sector_form.addRow("Sector:", self._sector_combo) filters_layout.addLayout(sector_form) # PE pe_form = QFormLayout() self._pe_min = QDoubleSpinBox() self._pe_min.setRange(0, 10000) self._pe_min.setSpecialValueText("—") self._pe_max = QDoubleSpinBox() self._pe_max.setRange(0, 10000) self._pe_max.setSpecialValueText("—") pe_form.addRow("P/E Min:", self._pe_min) pe_form.addRow("P/E Max:", self._pe_max) filters_layout.addLayout(pe_form) # Market Cap mc_form = QFormLayout() self._mc_min = QDoubleSpinBox() self._mc_min.setRange(0, 100000) self._mc_min.setSuffix("B") self._mc_min.setSpecialValueText("—") self._mc_max = QDoubleSpinBox() self._mc_max.setRange(0, 100000) self._mc_max.setSuffix("B") self._mc_max.setSpecialValueText("—") mc_form.addRow("MCap Min:", self._mc_min) mc_form.addRow("MCap Max:", self._mc_max) filters_layout.addLayout(mc_form) # RSI rsi_form = QFormLayout() self._rsi_min = QDoubleSpinBox() self._rsi_min.setRange(0, 100) self._rsi_min.setSpecialValueText("—") self._rsi_max = QDoubleSpinBox() self._rsi_max.setRange(0, 100) self._rsi_max.setValue(100) rsi_form.addRow("RSI Min:", self._rsi_min) rsi_form.addRow("RSI Max:", self._rsi_max) filters_layout.addLayout(rsi_form) # 52W week_form = QFormLayout() self._from_52h_min = QDoubleSpinBox() self._from_52h_min.setRange(-100, 0) self._from_52h_min.setSpecialValueText("—") self._from_52h_min.setSuffix("%") self._from_52h_max = QDoubleSpinBox() self._from_52h_max.setRange(-100, 100) self._from_52h_max.setSpecialValueText("—") self._from_52h_max.setSuffix("%") week_form.addRow("% from 52H min:", self._from_52h_min) week_form.addRow("% from 52H max:", self._from_52h_max) filters_layout.addLayout(week_form) root.addWidget(filters_box) # Run button + progress btn_row = QHBoxLayout() self._run_btn = QPushButton("Run Screener") self._run_btn.setObjectName("primary_btn") self._run_btn.clicked.connect(self._run_screener) self._progress = QProgressBar() self._progress.setMaximumWidth(200) self._progress.hide() btn_row.addWidget(self._run_btn) btn_row.addWidget(self._progress) btn_row.addStretch() self._result_label = QLabel("") self._result_label.setObjectName("subtitle") btn_row.addWidget(self._result_label) root.addLayout(btn_row) # Results table self._table = QTableWidget() self._table.setColumnCount(8) self._table.setHorizontalHeaderLabels(["Symbol", "Name", "Price", "Sector", "MCap", "P/E", "RSI", "% from 52H"]) self._table.setAlternatingRowColors(True) self._table.setSelectionBehavior(QTableWidget.SelectionBehavior.SelectRows) self._table.horizontalHeader().setStretchLastSection(True) self._table.doubleClicked.connect(self._on_row_double_click) self._table.setContextMenuPolicy(Qt.ContextMenuPolicy.CustomContextMenu) self._table.customContextMenuRequested.connect(self._show_context_menu) root.addWidget(self._table) def _run_screener(self): if self._worker and self._worker.isRunning(): return self._run_btn.setEnabled(False) self._progress.setValue(0) self._progress.show() self._result_label.setText("Scanning…") filters = { "sector": self._sector_combo.currentText(), "pe_min": self._pe_min.value() or None, "pe_max": self._pe_max.value() or None, "mc_min": self._mc_min.value() or None, "mc_max": self._mc_max.value() or None, "rsi_min": self._rsi_min.value() or None, "rsi_max": self._rsi_max.value() if self._rsi_max.value() < 100 else None, "from_52h_min": self._from_52h_min.value() or None, "from_52h_max": self._from_52h_max.value() or None, } self._worker = ScreenerWorker(filters) self._worker.data_ready.connect(self._on_results) self._worker.progress.connect(self._progress.setValue) self._worker.error.connect(lambda e: ( self._result_label.setText(f"Error: {e}"), self._run_btn.setEnabled(True), self._progress.hide(), )) self._worker.start() def _on_results(self, results: list[dict]): self._results = results self._table.setRowCount(len(results)) for row, r in enumerate(results): pct_52h = r.get("pct_from_52h") items_data = [ (r.get("symbol", ""), Qt.AlignmentFlag.AlignLeft), (r.get("name", "")[:30], Qt.AlignmentFlag.AlignLeft), (fmt_currency(r.get("price")), Qt.AlignmentFlag.AlignRight), (r.get("sector", ""), Qt.AlignmentFlag.AlignLeft), (fmt_large_number(r.get("market_cap")), Qt.AlignmentFlag.AlignRight), (f"{r['pe_ratio']:.1f}" if r.get("pe_ratio") else "—", Qt.AlignmentFlag.AlignRight), (f"{r['rsi']:.1f}" if r.get("rsi") else "—", Qt.AlignmentFlag.AlignRight), (fmt_percent(pct_52h, plus=True) if pct_52h is not None else "—", Qt.AlignmentFlag.AlignRight), ] for col, (text, align) in enumerate(items_data): cell = QTableWidgetItem(str(text)) cell.setTextAlignment(align | Qt.AlignmentFlag.AlignVCenter) if col == 7 and pct_52h is not None: cell.setForeground(QColor(pnl_color(pct_52h))) self._table.setItem(row, col, cell) self._run_btn.setEnabled(True) self._progress.hide() self._result_label.setText(f"{len(results)} results") def _on_row_double_click(self, index): row = index.row() if row < len(self._results): sym = self._results[row].get("symbol", "") if sym: self.symbol_selected.emit(sym) def _show_context_menu(self, pos): row = self._table.rowAt(pos.y()) if row < 0 or row >= len(self._results): return sym = self._results[row].get("symbol", "") menu = QMenu(self) chart_action = menu.addAction(f"View Chart: {sym}") watch_action = menu.addAction("Add to Watchlist") action = menu.exec(self._table.mapToGlobal(pos)) if action == chart_action: self.symbol_selected.emit(sym) elif action == watch_action: self.add_to_watchlist.emit(sym)