from __future__ import annotations import numpy as np import pandas as pd import pyqtgraph as pg from PyQt6.QtWidgets import ( QWidget, QVBoxLayout, QHBoxLayout, QLabel, QLineEdit, QPushButton, QCheckBox, QCompleter, QSizePolicy, QFrame ) from PyQt6.QtCore import Qt, QStringListModel, QThread, pyqtSignal, QTimer from PyQt6.QtGui import QPicture, QPainter, QColor, QPen, QBrush, QFont from PyQt6.QtCore import QRectF, QPointF from utils.formatters import fmt_currency, fmt_percent, pnl_color PERIODS = ["1D", "1W", "1M", "3M", "6M", "1Y", "5Y"] INDICATORS = ["SMA20", "SMA50", "EMA20", "BB", "Volume", "RSI", "MACD"] COLORS = { "up": "#a6e3a1", "down": "#f38ba8", "sma20": "#89b4fa", "sma50": "#fab387", "ema20": "#cba6f7", "bb_upper": "#89dceb", "bb_lower": "#89dceb", "bb_fill": "#1a3a4a", "volume_up": "#a6e3a1", "volume_down": "#f38ba8", "rsi_line": "#cba6f7", "macd_line": "#89b4fa", "macd_signal": "#fab387", "macd_pos": "#a6e3a1", "macd_neg": "#f38ba8", } class CandlestickItem(pg.GraphicsObject): def __init__(self, df: pd.DataFrame): super().__init__() self._df = df self._picture = QPicture() self._generate() def _generate(self): self._picture = QPicture() p = QPainter(self._picture) p.setRenderHint(QPainter.RenderHint.Antialiasing, False) w = 0.3 for i, (ts, row) in enumerate(self._df.iterrows()): t = float(i) o, c, lo, hi = float(row["Open"]), float(row["Close"]), float(row["Low"]), float(row["High"]) is_up = c >= o color = QColor(COLORS["up"] if is_up else COLORS["down"]) pen = QPen(color) pen.setWidthF(1.0) p.setPen(pen) p.setBrush(QBrush(color)) p.drawLine(QPointF(t, lo), QPointF(t, hi)) body_top = max(o, c) body_bot = min(o, c) body_h = max(body_top - body_bot, 0.001) p.drawRect(QRectF(t - w, body_bot, w * 2, body_h)) p.end() def paint(self, p, *args): p.drawPicture(0, 0, self._picture) def boundingRect(self): return QRectF(self._picture.boundingRect()) class DateAxisItem(pg.AxisItem): def __init__(self, timestamps: list, *args, **kwargs): super().__init__(*args, **kwargs) self._timestamps = timestamps def tickStrings(self, values, scale, spacing): result = [] for v in values: idx = int(round(v)) if 0 <= idx < len(self._timestamps): ts = self._timestamps[idx] if hasattr(ts, "strftime"): result.append(ts.strftime("%m/%d")) else: result.append(str(ts)[:10]) else: result.append("") return result class ChartDataWorker(QThread): data_ready = pyqtSignal(object, str) error = pyqtSignal(str) def __init__(self, symbol: str, period: str): super().__init__() self.symbol = symbol self.period = period def run(self): try: from core.market_data import get_chart_data df = get_chart_data(self.symbol, self.period) self.data_ready.emit(df, self.symbol) except Exception as e: self.error.emit(str(e)) class ChartWidget(QWidget): symbol_changed = pyqtSignal(str) def __init__(self, parent=None): super().__init__(parent) self._symbol = "AAPL" self._period = "6M" self._df: pd.DataFrame | None = None self._worker: ChartDataWorker | None = None self._active_indicators: set[str] = {"Volume"} self._setup_ui() self._setup_autocomplete() self.load_symbol(self._symbol) def _setup_ui(self): root = QVBoxLayout(self) root.setContentsMargins(12, 12, 12, 12) root.setSpacing(8) # Search + quote bar top_bar = QHBoxLayout() self._search = QLineEdit() self._search.setPlaceholderText("Search symbol… (e.g. AAPL, BTC-USD)") self._search.setMaximumWidth(280) self._search.returnPressed.connect(self._on_search) top_bar.addWidget(self._search) self._quote_label = QLabel() self._quote_label.setFont(QFont("Segoe UI", 11, QFont.Weight.Bold)) top_bar.addWidget(self._quote_label) self._change_label = QLabel() top_bar.addWidget(self._change_label) top_bar.addStretch() root.addLayout(top_bar) # Period buttons period_bar = QHBoxLayout() self._period_btns: dict[str, QPushButton] = {} for p in PERIODS: btn = QPushButton(p) btn.setCheckable(True) btn.setMaximumWidth(48) btn.clicked.connect(lambda checked, period=p: self._set_period(period)) self._period_btns[p] = btn period_bar.addWidget(btn) period_bar.addSpacing(20) for ind in INDICATORS: cb = QCheckBox(ind) cb.setChecked(ind == "Volume") cb.stateChanged.connect(lambda state, i=ind: self._toggle_indicator(i, state)) period_bar.addWidget(cb) period_bar.addStretch() root.addLayout(period_bar) self._period_btns["6M"].setChecked(True) # Chart area pg.setConfigOptions(antialias=True, foreground="#6c7086", background="#1e1e2e") self._layout_widget = pg.GraphicsLayoutWidget() root.addWidget(self._layout_widget) self._status_label = QLabel("Loading…") self._status_label.setObjectName("subtitle") root.addWidget(self._status_label) self._build_plots() def _build_plots(self): self._layout_widget.clear() self._main_plot: pg.PlotItem = self._layout_widget.addPlot(row=0, col=0) self._main_plot.showGrid(x=False, y=True, alpha=0.15) self._main_plot.getAxis("left").setWidth(60) self._main_plot.getAxis("bottom").setStyle(showValues=False) self._main_plot.setMinimumHeight(260) self._vol_plot: pg.PlotItem = self._layout_widget.addPlot(row=1, col=0) self._vol_plot.setMaximumHeight(80) self._vol_plot.showGrid(x=False, y=True, alpha=0.1) self._vol_plot.getAxis("left").setWidth(60) self._vol_plot.setXLink(self._main_plot) self._rsi_plot: pg.PlotItem = self._layout_widget.addPlot(row=2, col=0) self._rsi_plot.setMaximumHeight(100) self._rsi_plot.showGrid(x=False, y=True, alpha=0.1) self._rsi_plot.getAxis("left").setWidth(60) self._rsi_plot.setXLink(self._main_plot) self._rsi_plot.setYRange(0, 100) self._rsi_plot.hide() self._macd_plot: pg.PlotItem = self._layout_widget.addPlot(row=3, col=0) self._macd_plot.setMaximumHeight(100) self._macd_plot.showGrid(x=False, y=True, alpha=0.1) self._macd_plot.getAxis("left").setWidth(60) self._macd_plot.setXLink(self._main_plot) self._macd_plot.hide() def _setup_autocomplete(self): common = [ "AAPL", "MSFT", "GOOGL", "AMZN", "NVDA", "META", "TSLA", "SPY", "QQQ", "JPM", "BAC", "V", "MA", "NFLX", "AMD", "INTC", "DIS", "WMT", "KO", "BTC-USD", "ETH-USD", "SOL-USD", "BNB-USD", "XRP-USD", ] self._completer_model = QStringListModel(common) completer = QCompleter(self._completer_model) completer.setCaseSensitivity(Qt.CaseSensitivity.CaseInsensitive) self._search.setCompleter(completer) def _on_search(self): text = self._search.text().strip().upper() if text: self.load_symbol(text) def _set_period(self, period: str): for p, btn in self._period_btns.items(): btn.setChecked(p == period) self._period = period if self._symbol: self.load_symbol(self._symbol) def _toggle_indicator(self, indicator: str, state: int): if state: self._active_indicators.add(indicator) else: self._active_indicators.discard(indicator) if self._df is not None: self._render_chart(self._df) def load_symbol(self, symbol: str): self._symbol = symbol.upper() self._status_label.setText(f"Loading {self._symbol}…") self._quote_label.setText(self._symbol) self._change_label.clear() if self._worker and self._worker.isRunning(): self._worker.quit() self._worker = ChartDataWorker(self._symbol, self._period) self._worker.data_ready.connect(self._on_data_ready) self._worker.error.connect(lambda e: self._status_label.setText(f"Error: {e}")) self._worker.start() self.symbol_changed.emit(self._symbol) def _on_data_ready(self, df: pd.DataFrame, symbol: str): if df.empty: self._status_label.setText(f"No data for {symbol}") return self._df = df self._render_chart(df) self._status_label.setText(f"{symbol} | {len(df)} bars | {self._period}") self._update_quote_display(df) def _update_quote_display(self, df: pd.DataFrame): last = float(df["Close"].iloc[-1]) prev = float(df["Close"].iloc[-2]) if len(df) > 1 else last change = last - prev change_pct = (change / prev * 100) if prev else 0 self._quote_label.setText(f"{self._symbol} {fmt_currency(last)}") self._change_label.setText(f"{'+' if change >= 0 else ''}{fmt_currency(change)} {fmt_percent(change_pct, plus=True)}") self._change_label.setStyleSheet(f"color: {pnl_color(change)};") def _render_chart(self, df: pd.DataFrame): self._build_plots() xs = np.arange(len(df)) timestamps = list(df.index) date_axis = DateAxisItem(timestamps, orientation="bottom") self._main_plot.getAxis("bottom").hide() candle = CandlestickItem(df) self._main_plot.addItem(candle) # Overlaid indicators closes = df["Close"].values.astype(float) if "SMA20" in self._active_indicators and len(closes) >= 20: sma20 = pd.Series(closes).rolling(20).mean().values self._main_plot.plot(xs, sma20, pen=pg.mkPen(COLORS["sma20"], width=1.5), name="SMA20") if "SMA50" in self._active_indicators and len(closes) >= 50: sma50 = pd.Series(closes).rolling(50).mean().values self._main_plot.plot(xs, sma50, pen=pg.mkPen(COLORS["sma50"], width=1.5), name="SMA50") if "EMA20" in self._active_indicators and len(closes) >= 20: ema20 = pd.Series(closes).ewm(span=20).mean().values self._main_plot.plot(xs, ema20, pen=pg.mkPen(COLORS["ema20"], width=1.5), name="EMA20") if "BB" in self._active_indicators and len(closes) >= 20: sma = pd.Series(closes).rolling(20).mean() std = pd.Series(closes).rolling(20).std() upper = (sma + 2 * std).values lower = (sma - 2 * std).values self._main_plot.plot(xs, upper, pen=pg.mkPen(COLORS["bb_upper"], width=1, style=Qt.PenStyle.DashLine)) self._main_plot.plot(xs, lower, pen=pg.mkPen(COLORS["bb_lower"], width=1, style=Qt.PenStyle.DashLine)) fill = pg.FillBetweenItem( self._main_plot.plot(xs, upper, pen=None), self._main_plot.plot(xs, lower, pen=None), brush=pg.mkBrush(QColor(137, 220, 235, 30)), ) self._main_plot.addItem(fill) # Volume if "Volume" in self._active_indicators: volumes = df["Volume"].values.astype(float) colors = [COLORS["volume_up"] if df["Close"].iloc[i] >= df["Open"].iloc[i] else COLORS["volume_down"] for i in range(len(df))] brushes = [pg.mkBrush(QColor(c)) for c in colors] bar = pg.BarGraphItem(x=xs, height=volumes, width=0.6, brushes=brushes, pens=[pg.mkPen(None)] * len(xs)) self._vol_plot.addItem(bar) self._vol_plot.show() else: self._vol_plot.hide() # RSI if "RSI" in self._active_indicators and len(closes) >= 14: delta = pd.Series(closes).diff() gain = delta.clip(lower=0).rolling(14).mean() loss = (-delta.clip(upper=0)).rolling(14).mean() rs = gain / loss rsi = (100 - (100 / (1 + rs))).values self._rsi_plot.clear() self._rsi_plot.plot(xs, rsi, pen=pg.mkPen(COLORS["rsi_line"], width=1.5)) self._rsi_plot.addLine(y=70, pen=pg.mkPen("#f38ba8", width=1, style=Qt.PenStyle.DashLine)) self._rsi_plot.addLine(y=30, pen=pg.mkPen("#a6e3a1", width=1, style=Qt.PenStyle.DashLine)) self._rsi_plot.show() else: self._rsi_plot.hide() # MACD if "MACD" in self._active_indicators and len(closes) >= 26: ema12 = pd.Series(closes).ewm(span=12).mean() ema26 = pd.Series(closes).ewm(span=26).mean() macd_line = (ema12 - ema26).values signal_line = pd.Series(macd_line).ewm(span=9).mean().values histogram = macd_line - signal_line self._macd_plot.clear() hist_colors = [COLORS["macd_pos"] if v >= 0 else COLORS["macd_neg"] for v in histogram] bar2 = pg.BarGraphItem(x=xs, height=histogram, width=0.6, brushes=[pg.mkBrush(QColor(c)) for c in hist_colors], pens=[pg.mkPen(None)] * len(xs)) self._macd_plot.addItem(bar2) self._macd_plot.plot(xs, macd_line, pen=pg.mkPen(COLORS["macd_line"], width=1.5)) self._macd_plot.plot(xs, signal_line, pen=pg.mkPen(COLORS["macd_signal"], width=1.5)) self._macd_plot.show() else: self._macd_plot.hide()