263 lines
10 KiB
Python
263 lines
10 KiB
Python
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)
|