Mercurial
diff schwab_trader/dashboard.py @ 220:eb8b4230fdb9
[schwab-trader] Add guarded trading experiment
| author | MrJuneJune <me@mrjunejune.com> |
|---|---|
| date | Sun, 02 Aug 2026 08:52:13 -0700 |
| parents | |
| children |
line wrap: on
line diff
--- /dev/null Thu Jan 01 00:00:00 1970 +0000 +++ b/schwab_trader/dashboard.py Sun Aug 02 08:52:13 2026 -0700 @@ -0,0 +1,598 @@ +from __future__ import annotations + +import json +import os +import re +import sqlite3 +import time +from dataclasses import dataclass +from pathlib import Path +from typing import Any + +from schwab_trader.schwab_client import DEFAULT_TOKEN_FILE, SchwabConfig, load_tokens + + +DEFAULT_DASHBOARD_DB = "~/.local/share/zenbu/schwab_trader/dashboard.db" + +DEFAULT_SETTINGS: dict[str, Any] = { + "profit_target_pct": 3.0, + "stop_loss_pct": 2.0, + "max_trade_dollars": 500.0, + "max_account_pct": 2.0, + "max_open_positions": 3, + "min_confidence": 0.60, + "min_evidence_count": 3, + "min_source_count": 2, + "cooldown_minutes": 60, + "market_hours_only": True, + "max_daily_loss_dollars": 250.0, + "day_trade_limit": 3, + "live_trading_enabled": False, + "require_manual_confirmation": True, + "llm_enabled": False, + "allowlist": [], + "blocklist": [], +} + +POSITIVE_WORDS = { + "beat", + "beats", + "bull", + "bullish", + "buy", + "calls", + "growth", + "hype", + "moon", + "mooning", + "profit", + "rally", + "strong", + "surge", + "up", + "winner", +} + +NEGATIVE_WORDS = { + "bear", + "bearish", + "crash", + "dump", + "fall", + "falling", + "fraud", + "lawsuit", + "loss", + "miss", + "puts", + "risk", + "sell", + "short", + "weak", +} + +SYMBOL_RE = re.compile(r"(?<![A-Z0-9])\$?([A-Z]{1,5})(?![A-Z0-9])") +COMMON_WORDS = { + "A", + "AI", + "AM", + "API", + "CEO", + "CFO", + "DD", + "ETF", + "GDP", + "IPO", + "IRS", + "LLM", + "PDT", + "SEC", + "USA", + "USD", +} + + +@dataclass(frozen=True) +class EvidenceInput: + source: str + text: str + symbol: str | None = None + url: str | None = None + engagement: float = 0.0 + raw: dict[str, Any] | None = None + + +class DashboardStore: + def __init__(self, db_path: Path | str | None = None) -> None: + if db_path is None: + db_path = os.environ.get("SCHWAB_DASHBOARD_DB", DEFAULT_DASHBOARD_DB) + self.db_path = Path(db_path).expanduser() + self.db_path.parent.mkdir(parents=True, exist_ok=True) + self._init_db() + + def get_status(self) -> dict[str, Any]: + token_file = Path(os.environ.get("SCHWAB_TOKEN_FILE", DEFAULT_TOKEN_FILE)).expanduser() + token_status: dict[str, Any] = { + "path": str(token_file), + "exists": token_file.exists(), + "access_token_present": False, + "refresh_token_present": False, + "saved_at": None, + "age_seconds": None, + } + if token_file.exists(): + try: + tokens = load_tokens(token_file) + saved_at = tokens.get("saved_at") + token_status.update( + { + "access_token_present": bool(tokens.get("access_token")), + "refresh_token_present": bool(tokens.get("refresh_token")), + "saved_at": saved_at, + "age_seconds": int(time.time()) - int(saved_at) if saved_at else None, + } + ) + except (OSError, ValueError, TypeError) as error: + token_status["error"] = str(error) + + env_status = { + "SCHWAB_APP_KEY": bool(os.environ.get("SCHWAB_APP_KEY")), + "SCHWAB_APP_SECRET": bool(os.environ.get("SCHWAB_APP_SECRET")), + "SCHWAB_REDIRECT_URI": bool(os.environ.get("SCHWAB_REDIRECT_URI")), + } + + return { + "service": "schwab-dashboard", + "database": str(self.db_path), + "env": env_status, + "tokens": token_status, + "live_trading_enabled": False, + "live_trading_note": "Dashboard has no live-trade endpoint; use CLI dry-run/manual confirmation flow.", + "counts": self.get_counts(), + } + + def get_counts(self) -> dict[str, int]: + with self._connect() as conn: + return { + "evidence": self._count(conn, "evidence"), + "signals": self._count(conn, "signals"), + "paper_trades": self._count(conn, "paper_trades"), + "audit_events": self._count(conn, "audit"), + } + + def get_settings(self) -> dict[str, Any]: + settings = dict(DEFAULT_SETTINGS) + with self._connect() as conn: + for row in conn.execute("SELECT key, value_json FROM settings"): + settings[row["key"]] = json.loads(row["value_json"]) + return settings + + def update_settings(self, updates: dict[str, Any]) -> dict[str, Any]: + allowed = set(DEFAULT_SETTINGS) + unknown = sorted(set(updates) - allowed) + if unknown: + raise ValueError("Unknown settings: " + ", ".join(unknown)) + + current = self.get_settings() + current.update(updates) + self._validate_settings(current) + + with self._connect() as conn: + for key, value in current.items(): + conn.execute( + """ + INSERT INTO settings(key, value_json) + VALUES (?, ?) + ON CONFLICT(key) DO UPDATE SET value_json = excluded.value_json + """, + (key, json.dumps(value, sort_keys=True)), + ) + conn.commit() + self.add_audit("settings.updated", "Dashboard settings updated", updates) + return current + + def add_evidence(self, item: EvidenceInput) -> dict[str, Any]: + source = item.source.strip().lower() + text = item.text.strip() + if not source: + raise ValueError("source is required") + if not text: + raise ValueError("text is required") + + symbol = normalize_symbol(item.symbol) if item.symbol else extract_symbol(text) + if not symbol: + raise ValueError("symbol is required or must be detectable as a ticker in text") + + sentiment_score = score_sentiment(text) + created_at = int(time.time()) + + with self._connect() as conn: + cursor = conn.execute( + """ + INSERT INTO evidence( + source, symbol, url, text, engagement, sentiment_score, raw_json, created_at + ) + VALUES (?, ?, ?, ?, ?, ?, ?, ?) + """, + ( + source, + symbol, + item.url, + text, + float(item.engagement), + sentiment_score, + json.dumps(item.raw or {}, sort_keys=True), + created_at, + ), + ) + evidence_id = int(cursor.lastrowid) + conn.commit() + + signal = self.recompute_signal(symbol) + self.add_audit( + "evidence.added", + f"Added {source} evidence for {symbol}", + {"evidence_id": evidence_id, "symbol": symbol, "signal": signal}, + ) + return {"id": evidence_id, "symbol": symbol, "sentiment_score": sentiment_score, "signal": signal} + + def list_evidence(self, limit: int = 100) -> list[dict[str, Any]]: + with self._connect() as conn: + rows = conn.execute( + """ + SELECT id, source, symbol, url, text, engagement, sentiment_score, created_at + FROM evidence + ORDER BY id DESC + LIMIT ? + """, + (limit,), + ).fetchall() + return [dict(row) for row in rows] + + def recompute_signal(self, symbol: str) -> dict[str, Any]: + symbol = normalize_symbol(symbol) + settings = self.get_settings() + with self._connect() as conn: + rows = conn.execute( + """ + SELECT source, sentiment_score, engagement, created_at + FROM evidence + WHERE symbol = ? + ORDER BY id DESC + LIMIT 100 + """, + (symbol,), + ).fetchall() + + evidence_count = len(rows) + source_count = len({row["source"] for row in rows}) + weighted_total = 0.0 + weight_sum = 0.0 + for row in rows: + engagement_weight = min(5.0, 1.0 + max(0.0, float(row["engagement"])) / 100.0) + weighted_total += float(row["sentiment_score"]) * engagement_weight + weight_sum += engagement_weight + sentiment_score = weighted_total / weight_sum if weight_sum else 0.0 + confidence = compute_confidence(sentiment_score, evidence_count, source_count) + action = decide_signal_action(symbol, sentiment_score, confidence, evidence_count, source_count, settings) + summary = summarize_signal(symbol, sentiment_score, confidence, evidence_count, source_count, action) + updated_at = int(time.time()) + + conn.execute( + """ + INSERT INTO signals( + symbol, sentiment_score, confidence, evidence_count, source_count, + action, summary, updated_at + ) + VALUES (?, ?, ?, ?, ?, ?, ?, ?) + ON CONFLICT(symbol) DO UPDATE SET + sentiment_score = excluded.sentiment_score, + confidence = excluded.confidence, + evidence_count = excluded.evidence_count, + source_count = excluded.source_count, + action = excluded.action, + summary = excluded.summary, + updated_at = excluded.updated_at + """, + ( + symbol, + sentiment_score, + confidence, + evidence_count, + source_count, + action, + summary, + updated_at, + ), + ) + conn.commit() + + return { + "symbol": symbol, + "sentiment_score": round(sentiment_score, 4), + "confidence": round(confidence, 4), + "evidence_count": evidence_count, + "source_count": source_count, + "action": action, + "summary": summary, + "updated_at": updated_at, + } + + def list_signals(self) -> list[dict[str, Any]]: + with self._connect() as conn: + rows = conn.execute( + """ + SELECT symbol, sentiment_score, confidence, evidence_count, source_count, + action, summary, updated_at + FROM signals + ORDER BY confidence DESC, updated_at DESC + """ + ).fetchall() + return [dict(row) for row in rows] + + def add_paper_trade(self, payload: dict[str, Any]) -> dict[str, Any]: + symbol = normalize_symbol(str(payload.get("symbol", ""))) + action = str(payload.get("action", "")).upper() + quantity = float(payload.get("quantity", 0)) + price = float(payload.get("price", 0)) + reason = str(payload.get("reason", "manual paper trade")).strip() + + if action not in {"BUY", "SELL"}: + raise ValueError("action must be BUY or SELL") + if not symbol: + raise ValueError("symbol is required") + if quantity <= 0: + raise ValueError("quantity must be greater than zero") + if price <= 0: + raise ValueError("price must be greater than zero") + + settings = self.get_settings() + notional = quantity * price + status = "accepted" + if notional > float(settings["max_trade_dollars"]): + status = "rejected_max_trade_dollars" + elif symbol in {normalize_symbol(s) for s in settings["blocklist"]}: + status = "rejected_blocklist" + + created_at = int(time.time()) + with self._connect() as conn: + cursor = conn.execute( + """ + INSERT INTO paper_trades(symbol, action, quantity, price, notional, status, reason, created_at) + VALUES (?, ?, ?, ?, ?, ?, ?, ?) + """, + (symbol, action, quantity, price, notional, status, reason, created_at), + ) + trade_id = int(cursor.lastrowid) + conn.commit() + + result = { + "id": trade_id, + "symbol": symbol, + "action": action, + "quantity": quantity, + "price": price, + "notional": notional, + "status": status, + "reason": reason, + "created_at": created_at, + } + self.add_audit("paper_trade.created", f"Paper trade {status}: {action} {quantity} {symbol}", result) + return result + + def list_paper_trades(self, limit: int = 100) -> list[dict[str, Any]]: + with self._connect() as conn: + rows = conn.execute( + """ + SELECT id, symbol, action, quantity, price, notional, status, reason, created_at + FROM paper_trades + ORDER BY id DESC + LIMIT ? + """, + (limit,), + ).fetchall() + return [dict(row) for row in rows] + + def add_audit(self, event_type: str, message: str, payload: dict[str, Any] | None = None) -> None: + with self._connect() as conn: + conn.execute( + """ + INSERT INTO audit(event_type, message, payload_json, created_at) + VALUES (?, ?, ?, ?) + """, + (event_type, message, json.dumps(payload or {}, sort_keys=True), int(time.time())), + ) + conn.commit() + + def list_audit(self, limit: int = 100) -> list[dict[str, Any]]: + with self._connect() as conn: + rows = conn.execute( + """ + SELECT id, event_type, message, payload_json, created_at + FROM audit + ORDER BY id DESC + LIMIT ? + """, + (limit,), + ).fetchall() + events = [] + for row in rows: + event = dict(row) + event["payload"] = json.loads(event.pop("payload_json")) + events.append(event) + return events + + def _init_db(self) -> None: + with self._connect() as conn: + conn.executescript( + """ + CREATE TABLE IF NOT EXISTS settings ( + key TEXT PRIMARY KEY, + value_json TEXT NOT NULL + ); + + CREATE TABLE IF NOT EXISTS evidence ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + source TEXT NOT NULL, + symbol TEXT NOT NULL, + url TEXT, + text TEXT NOT NULL, + engagement REAL NOT NULL DEFAULT 0, + sentiment_score REAL NOT NULL, + raw_json TEXT NOT NULL DEFAULT '{}', + created_at INTEGER NOT NULL + ); + + CREATE INDEX IF NOT EXISTS idx_evidence_symbol_created + ON evidence(symbol, created_at); + + CREATE TABLE IF NOT EXISTS signals ( + symbol TEXT PRIMARY KEY, + sentiment_score REAL NOT NULL, + confidence REAL NOT NULL, + evidence_count INTEGER NOT NULL, + source_count INTEGER NOT NULL, + action TEXT NOT NULL, + summary TEXT NOT NULL, + updated_at INTEGER NOT NULL + ); + + CREATE TABLE IF NOT EXISTS paper_trades ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + symbol TEXT NOT NULL, + action TEXT NOT NULL, + quantity REAL NOT NULL, + price REAL NOT NULL, + notional REAL NOT NULL, + status TEXT NOT NULL, + reason TEXT NOT NULL, + created_at INTEGER NOT NULL + ); + + CREATE TABLE IF NOT EXISTS audit ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + event_type TEXT NOT NULL, + message TEXT NOT NULL, + payload_json TEXT NOT NULL, + created_at INTEGER NOT NULL + ); + """ + ) + for key, value in DEFAULT_SETTINGS.items(): + conn.execute( + "INSERT OR IGNORE INTO settings(key, value_json) VALUES (?, ?)", + (key, json.dumps(value, sort_keys=True)), + ) + conn.commit() + + def _connect(self) -> sqlite3.Connection: + conn = sqlite3.connect(self.db_path) + conn.row_factory = sqlite3.Row + return conn + + @staticmethod + def _count(conn: sqlite3.Connection, table: str) -> int: + return int(conn.execute(f"SELECT COUNT(*) FROM {table}").fetchone()[0]) + + @staticmethod + def _validate_settings(settings: dict[str, Any]) -> None: + positive_numbers = [ + "profit_target_pct", + "stop_loss_pct", + "max_trade_dollars", + "max_account_pct", + "max_daily_loss_dollars", + ] + for key in positive_numbers: + if float(settings[key]) <= 0: + raise ValueError(f"{key} must be greater than zero") + if not 0 <= float(settings["min_confidence"]) <= 1: + raise ValueError("min_confidence must be between 0 and 1") + if int(settings["min_evidence_count"]) < 1: + raise ValueError("min_evidence_count must be at least 1") + if int(settings["min_source_count"]) < 1: + raise ValueError("min_source_count must be at least 1") + if bool(settings["live_trading_enabled"]): + raise ValueError("live_trading_enabled cannot be enabled from the dashboard") + + +def normalize_symbol(value: str) -> str: + symbol = value.strip().upper().lstrip("$") + if not re.fullmatch(r"[A-Z]{1,5}", symbol): + raise ValueError("symbol must be 1-5 letters") + return symbol + + +def extract_symbol(text: str) -> str | None: + for match in SYMBOL_RE.finditer(text.upper()): + symbol = match.group(1) + if symbol not in COMMON_WORDS: + return symbol + return None + + +def score_sentiment(text: str) -> float: + words = re.findall(r"[a-zA-Z']+", text.lower()) + positive = sum(1 for word in words if word in POSITIVE_WORDS) + negative = sum(1 for word in words if word in NEGATIVE_WORDS) + total = positive + negative + if total == 0: + return 0.0 + return max(-1.0, min(1.0, (positive - negative) / total)) + + +def compute_confidence(sentiment_score: float, evidence_count: int, source_count: int) -> float: + evidence_component = min(0.35, evidence_count * 0.07) + source_component = min(0.25, source_count * 0.10) + sentiment_component = min(0.20, abs(sentiment_score) * 0.20) + return min(0.95, 0.20 + evidence_component + source_component + sentiment_component) + + +def decide_signal_action( + symbol: str, + sentiment_score: float, + confidence: float, + evidence_count: int, + source_count: int, + settings: dict[str, Any], +) -> str: + blocklist = {normalize_symbol(item) for item in settings.get("blocklist", [])} + allowlist = {normalize_symbol(item) for item in settings.get("allowlist", [])} + if symbol in blocklist: + return "NO_TRADE_BLOCKED" + if allowlist and symbol not in allowlist: + return "NO_TRADE_NOT_ALLOWLISTED" + if evidence_count < int(settings["min_evidence_count"]): + return "NO_TRADE_NEEDS_EVIDENCE" + if source_count < int(settings["min_source_count"]): + return "NO_TRADE_NEEDS_SOURCE_DIVERSITY" + if confidence < float(settings["min_confidence"]): + return "NO_TRADE_LOW_CONFIDENCE" + if sentiment_score >= 0.25: + return "CONSIDER_BUY" + if sentiment_score <= -0.25: + return "CONSIDER_SELL" + return "WATCH" + + +def summarize_signal( + symbol: str, + sentiment_score: float, + confidence: float, + evidence_count: int, + source_count: int, + action: str, +) -> str: + direction = "positive" if sentiment_score > 0 else "negative" if sentiment_score < 0 else "neutral" + return ( + f"{symbol} has {direction} social sentiment from {evidence_count} evidence item(s) " + f"across {source_count} source(s). Confidence is {confidence:.0%}. Action: {action}." + ) + + +def get_config_if_available() -> SchwabConfig | None: + try: + return SchwabConfig.from_env() + except Exception: + return None +