438 lines
17 KiB
Python
438 lines
17 KiB
Python
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()
|