from __future__ import annotations from PyQt6.QtWidgets import ( QWidget, QVBoxLayout, QHBoxLayout, QLabel, QPushButton, QTextEdit, QTabWidget, QLineEdit, QSplitter, QFrame, QMessageBox ) from PyQt6.QtCore import Qt, QThread, pyqtSignal from PyQt6.QtGui import QFont, QTextCursor class AIWorker(QThread): result_ready = pyqtSignal(str, dict) error = pyqtSignal(str) def __init__(self, func, *args, **kwargs): super().__init__() self._func = func self._args = args self._kwargs = kwargs def run(self): try: text, usage = self._func(*self._args, **self._kwargs) self.result_ready.emit(text, usage) except Exception as e: self.error.emit(str(e)) class AIResultWidget(QFrame): def __init__(self, parent=None): super().__init__(parent) self.setObjectName("card") layout = QVBoxLayout(self) layout.setContentsMargins(8, 8, 8, 8) self._text = QTextEdit() self._text.setReadOnly(True) self._text.setFont(QFont("Segoe UI", 10)) layout.addWidget(self._text) self._cost_label = QLabel("") self._cost_label.setObjectName("subtitle") layout.addWidget(self._cost_label) def set_text(self, text: str): self._text.setPlainText(text) cursor = self._text.textCursor() cursor.movePosition(QTextCursor.MoveOperation.Start) self._text.setTextCursor(cursor) def set_loading(self): self._text.setPlainText("Analyzing… please wait.") self._cost_label.clear() def set_error(self, error: str): self._text.setPlainText(f"Error: {error}") self._cost_label.clear() def set_usage(self, usage: dict): if usage: in_tok = usage.get("input_tokens", 0) out_tok = usage.get("output_tokens", 0) cost = usage.get("cost_usd", 0) self._cost_label.setText(f"Tokens: {in_tok} in / {out_tok} out | Est. cost: ${cost:.4f}") class AIPanelWidget(QWidget): def __init__(self, config, parent=None): super().__init__(parent) self._config = config self._symbol = "AAPL" self._worker: AIWorker | None = None self._setup_ui() def _setup_ui(self): root = QVBoxLayout(self) root.setContentsMargins(12, 12, 12, 12) root.setSpacing(8) header_row = QHBoxLayout() header = QLabel("AI Analysis") header.setObjectName("title") header_row.addWidget(header) self._symbol_label = QLabel(f"Analyzing: {self._symbol}") self._symbol_label.setObjectName("subtitle") header_row.addWidget(self._symbol_label) header_row.addStretch() root.addLayout(header_row) self._tabs = QTabWidget() self._tabs.addTab(self._build_summary_tab(), "Summary") self._tabs.addTab(self._build_technical_tab(), "Technical") self._tabs.addTab(self._build_sentiment_tab(), "Sentiment") self._tabs.addTab(self._build_outlook_tab(), "Outlook") self._tabs.addTab(self._build_chat_tab(), "Chat") self._tabs.addTab(self._build_portfolio_tab(), "Portfolio") root.addWidget(self._tabs) def _make_tab_layout(self, title: str, btn_text: str, on_click) -> tuple[QWidget, AIResultWidget]: tab = QWidget() layout = QVBoxLayout(tab) layout.setContentsMargins(8, 8, 8, 8) top_row = QHBoxLayout() lbl = QLabel(title) lbl.setObjectName("subtitle") top_row.addWidget(lbl) top_row.addStretch() btn = QPushButton(btn_text) btn.setObjectName("primary_btn") btn.setMaximumWidth(160) btn.clicked.connect(on_click) top_row.addWidget(btn) layout.addLayout(top_row) result = AIResultWidget() layout.addWidget(result) return tab, result def _build_summary_tab(self) -> QWidget: tab, self._summary_result = self._make_tab_layout( "AI Stock Summary — fundamentals + news digest", "Run Summary", self._run_summary, ) return tab def _build_technical_tab(self) -> QWidget: tab, self._technical_result = self._make_tab_layout( "AI Technical Read — chart pattern analysis", "Run Technical", self._run_technical, ) return tab def _build_sentiment_tab(self) -> QWidget: tab, self._sentiment_result = self._make_tab_layout( "AI Sentiment Score — news → Bull/Bear/Neutral", "Run Sentiment", self._run_sentiment, ) return tab def _build_outlook_tab(self) -> QWidget: tab, self._outlook_result = self._make_tab_layout( "AI Price Outlook — 1D/1W/1M directional bias", "Run Outlook", self._run_outlook, ) return tab def _build_chat_tab(self) -> QWidget: tab = QWidget() layout = QVBoxLayout(tab) layout.setContentsMargins(8, 8, 8, 8) self._chat_display = QTextEdit() self._chat_display.setReadOnly(True) self._chat_display.setFont(QFont("Segoe UI", 10)) layout.addWidget(self._chat_display, stretch=1) self._chat_cost_label = QLabel("") self._chat_cost_label.setObjectName("subtitle") layout.addWidget(self._chat_cost_label) input_row = QHBoxLayout() self._chat_input = QLineEdit() self._chat_input.setPlaceholderText(f"Ask anything about {self._symbol}…") self._chat_input.returnPressed.connect(self._send_chat) send_btn = QPushButton("Send") send_btn.setObjectName("primary_btn") send_btn.clicked.connect(self._send_chat) input_row.addWidget(self._chat_input) input_row.addWidget(send_btn) layout.addLayout(input_row) return tab def _build_portfolio_tab(self) -> QWidget: tab, self._portfolio_result = self._make_tab_layout( "AI Portfolio Review — risks, correlation, recommendations", "Review Portfolio", self._run_portfolio_review, ) return tab def _get_client(self): from ai.claude_client import ClaudeClient key = self._config.anthropic_key if not key: raise RuntimeError("Anthropic API key not configured. Go to Settings.") return ClaudeClient(key) def set_symbol(self, symbol: str): self._symbol = symbol self._symbol_label.setText(f"Analyzing: {symbol}") self._chat_input.setPlaceholderText(f"Ask anything about {symbol}…") def _run_summary(self): self._summary_result.set_loading() try: client = self._get_client() except RuntimeError as e: self._summary_result.set_error(str(e)) return def _fetch(): from core.market_data import get_fundamentals, get_quote from ai.sentiment import fetch_news from ai.prompts import STOCK_SUMMARY_SYSTEM, STOCK_SUMMARY_PROMPT from utils.formatters import fmt_currency, fmt_large_number info = get_fundamentals(self._symbol) news = fetch_news(self._symbol, self._config.news_api_key, page_size=5) headlines = "\n".join(f"- {a['title']}" for a in news[:5]) or "No recent news." prompt = STOCK_SUMMARY_PROMPT.format( symbol=self._symbol, name=info.get("name", self._symbol), sector=info.get("sector", "N/A"), industry=info.get("industry", "N/A"), market_cap=fmt_large_number(info.get("market_cap")), pe_ratio=f"{info.get('pe_ratio', 'N/A')}", eps=f"{info.get('eps', 'N/A')}", week_52_high=fmt_currency(info.get("week_52_high")), week_52_low=fmt_currency(info.get("week_52_low")), dividend_yield=f"{info.get('dividend_yield', 'N/A')}", headlines=headlines, ) return client.ask(prompt, system=STOCK_SUMMARY_SYSTEM, max_tokens=1024) self._worker = AIWorker(_fetch) self._worker.result_ready.connect(lambda t, u: (self._summary_result.set_text(t), self._summary_result.set_usage(u))) self._worker.error.connect(self._summary_result.set_error) self._worker.start() def _run_technical(self): self._technical_result.set_loading() try: client = self._get_client() except RuntimeError as e: self._technical_result.set_error(str(e)) return def _fetch(): import numpy as np import pandas as pd from core.market_data import get_chart_data from ai.prompts import TECHNICAL_READ_SYSTEM, TECHNICAL_READ_PROMPT from utils.formatters import fmt_currency df = get_chart_data(self._symbol, "3M") if df.empty: raise RuntimeError("No chart data available.") closes = df["Close"].values.astype(float) current_price = closes[-1] sma20 = float(pd.Series(closes).rolling(20).mean().iloc[-1]) if len(closes) >= 20 else None sma50 = float(pd.Series(closes).rolling(50).mean().iloc[-1]) if len(closes) >= 50 else None delta = pd.Series(closes).diff() gain = delta.clip(lower=0).rolling(14).mean() loss = (-delta.clip(upper=0)).rolling(14).mean() rsi_val = float(100 - (100 / (1 + gain / loss)).iloc[-1]) if len(closes) >= 14 else None ema12 = pd.Series(closes).ewm(span=12).mean() ema26 = pd.Series(closes).ewm(span=26).mean() macd_val = float((ema12 - ema26).iloc[-1]) signal_val = float((ema12 - ema26).ewm(span=9).mean().iloc[-1]) sma20_s = pd.Series(closes).rolling(20) bb_upper = float((sma20_s.mean() + 2 * sma20_s.std()).iloc[-1]) if len(closes) >= 20 else None bb_lower = float((sma20_s.mean() - 2 * sma20_s.std()).iloc[-1]) if len(closes) >= 20 else None trend = "uptrend" if len(closes) >= 20 and closes[-1] > closes[-20] else "downtrend" prompt = TECHNICAL_READ_PROMPT.format( symbol=self._symbol, period="3M", price=fmt_currency(current_price), sma20=fmt_currency(sma20) if sma20 else "N/A", sma50=fmt_currency(sma50) if sma50 else "N/A", rsi=f"{rsi_val:.1f}" if rsi_val else "N/A", macd=f"{macd_val:.3f}", macd_signal=f"{signal_val:.3f}", bb_upper=fmt_currency(bb_upper) if bb_upper else "N/A", bb_lower=fmt_currency(bb_lower) if bb_lower else "N/A", price_action=trend, ) return client.ask(prompt, system=TECHNICAL_READ_SYSTEM, max_tokens=800) self._worker = AIWorker(_fetch) self._worker.result_ready.connect(lambda t, u: (self._technical_result.set_text(t), self._technical_result.set_usage(u))) self._worker.error.connect(self._technical_result.set_error) self._worker.start() def _run_sentiment(self): self._sentiment_result.set_loading() try: client = self._get_client() except RuntimeError as e: self._sentiment_result.set_error(str(e)) return def _fetch(): from ai.sentiment import fetch_news, analyze_sentiment articles = fetch_news(self._symbol, self._config.news_api_key) return analyze_sentiment(self._symbol, articles, client) self._worker = AIWorker(_fetch) self._worker.result_ready.connect(lambda t, u: (self._sentiment_result.set_text(t), self._sentiment_result.set_usage(u))) self._worker.error.connect(self._sentiment_result.set_error) self._worker.start() def _run_outlook(self): self._outlook_result.set_loading() try: client = self._get_client() except RuntimeError as e: self._outlook_result.set_error(str(e)) return def _fetch(): import pandas as pd from core.market_data import get_chart_data, get_quote from ai.prompts import PRICE_OUTLOOK_SYSTEM, PRICE_OUTLOOK_PROMPT from utils.formatters import fmt_currency df = get_chart_data(self._symbol, "1M") closes = df["Close"].values.astype(float) if not df.empty else [] rsi_val = "N/A" trend = "neutral" momentum = "neutral" if len(closes) >= 14: delta = pd.Series(closes).diff() gain = delta.clip(lower=0).rolling(14).mean() loss = (-delta.clip(upper=0)).rolling(14).mean() rsi_v = float(100 - (100 / (1 + gain / loss)).iloc[-1]) rsi_val = f"{rsi_v:.1f}" trend = "uptrend" if closes[-1] > closes[-5] else "downtrend" momentum = "bullish" if rsi_v > 55 else ("bearish" if rsi_v < 45 else "neutral") quote = get_quote(self._symbol) prompt = PRICE_OUTLOOK_PROMPT.format( symbol=self._symbol, price=fmt_currency(quote.get("price", 0)), rsi=rsi_val, trend=trend, momentum=momentum, sentiment="N/A", sector_perf="N/A", ) return client.ask(prompt, system=PRICE_OUTLOOK_SYSTEM, max_tokens=600) self._worker = AIWorker(_fetch) self._worker.result_ready.connect(lambda t, u: (self._outlook_result.set_text(t), self._outlook_result.set_usage(u))) self._worker.error.connect(self._outlook_result.set_error) self._worker.start() def _send_chat(self): question = self._chat_input.text().strip() if not question: return try: client = self._get_client() except RuntimeError as e: self._chat_display.append(f"\nError: {e}") return self._chat_display.append(f"\nYou: {question}") self._chat_input.clear() self._chat_display.append("AI: thinking…\n") def _fetch(): from ai.prompts import CHAT_SYSTEM context = f"Current ticker being analyzed: {self._symbol}\nUser question: {question}" return client.ask(context, system=CHAT_SYSTEM, max_tokens=1024) def _on_result(text: str, usage: dict): cursor = self._chat_display.textCursor() cursor.movePosition(QTextCursor.MoveOperation.End) cursor.movePosition(QTextCursor.MoveOperation.StartOfBlock, QTextCursor.MoveMode.KeepAnchor) cursor.removeSelectedText() self._chat_display.append(f"AI: {text}\n") in_tok = usage.get("input_tokens", 0) out_tok = usage.get("output_tokens", 0) cost = usage.get("cost_usd", 0) self._chat_cost_label.setText(f"Last: {in_tok}in/{out_tok}out — ${cost:.4f}") self._worker = AIWorker(_fetch) self._worker.result_ready.connect(_on_result) self._worker.error.connect(lambda e: self._chat_display.append(f"Error: {e}\n")) self._worker.start() def _run_portfolio_review(self): self._portfolio_result.set_loading() try: client = self._get_client() except RuntimeError as e: self._portfolio_result.set_error(str(e)) return def _fetch(): from db.database import get_session from db.models import PortfolioPosition from core.market_data import get_batch_quotes from core.portfolio import calculate_pnl, calculate_sector_allocation from ai.prompts import PORTFOLIO_REVIEW_SYSTEM, PORTFOLIO_REVIEW_PROMPT from utils.formatters import fmt_currency, fmt_percent with get_session() as session: rows = session.query(PortfolioPosition).all() raw = [{"symbol": r.symbol, "shares": r.shares, "avg_cost": r.avg_cost} for r in rows] if not raw: raise RuntimeError("No portfolio positions found. Add positions first.") quotes = get_batch_quotes([r["symbol"] for r in raw]) positions = calculate_pnl(raw, quotes) total_value = sum(p.get("market_value", 0) for p in positions) cash = 0.0 top_pos = max(positions, key=lambda p: p.get("market_value", 0), default={}) sector_alloc = calculate_sector_allocation(positions) holdings_str = "\n".join( f"- {p['symbol']}: {p.get('shares', 0):.2f} shares @ {fmt_currency(p.get('avg_cost', 0))}, " f"value={fmt_currency(p.get('market_value', 0))}, P&L={fmt_percent(p.get('pnl_pct', 0))}" for p in positions ) sector_str = ", ".join(f"{s}: {v:.1f}%" for s, v in sector_alloc.items()) prompt = PORTFOLIO_REVIEW_PROMPT.format( holdings=holdings_str, total_value=fmt_currency(total_value), cash=fmt_currency(cash), top_position=top_pos.get("symbol", "N/A"), sector_allocation=sector_str, ) return client.ask(prompt, system=PORTFOLIO_REVIEW_SYSTEM, max_tokens=1200) self._worker = AIWorker(_fetch) self._worker.result_ready.connect(lambda t, u: (self._portfolio_result.set_text(t), self._portfolio_result.set_usage(u))) self._worker.error.connect(self._portfolio_result.set_error) self._worker.start()