diff --git a/.env.example b/.env.example index d04e409..94782bd 100644 --- a/.env.example +++ b/.env.example @@ -61,6 +61,14 @@ Alpha_Vantage_Api_Key= # dilution veto — block confirmed dilutive offerings before sizing. # NEWS_DILUTION_VETO_ENABLED=0 # NEWS_DILUTION_VETO_CONVICTION=0.6 +# +# AI trade analysis (advisory): the local model reads the armed/weak setups, the +# day's closed trades, and the whole session, writing a pursue/avoid verdict + +# post-mortems + an EOD note to ai_trade_analysis_cache (shown on the dashboard). +# OFF the hot path; never gates a trade. Needs OLLAMA_ENABLED=1. +# TRADE_ANALYSIS_ENABLED=0 +# TRADE_ANALYSIS_INTERVAL_SECONDS=180 +# TRADE_ANALYSIS_BATCH_LIMIT=8 # max setups analyzed per pass (GPU budget) # ============================================================ # Live paper-trading orchestrator (run_live_paper.py) diff --git a/api/main.py b/api/main.py index 0a75563..f56d5e6 100644 --- a/api/main.py +++ b/api/main.py @@ -42,6 +42,7 @@ from storage.projections import ( query_order_lifecycle_snapshot, query_ready_signals_snapshot, query_risk_state, + query_trade_analysis, query_session_pnl, query_session_summary, query_symbol_criteria, @@ -104,11 +105,26 @@ class DashboardState: # Local-LLM catalyst advisory (live only). {} when enrichment has # never run, so the dashboard simply shows no catalyst column. catalyst = query_catalyst_advisory(store) if is_live else {} + # Local-LLM AI trade analysis (armed/weak verdict, post-mortems, + # EOD note). Works live and for a past session; {} until it runs. + trade_analysis = query_trade_analysis(store, for_date=for_date) + armed_verdicts = trade_analysis.get("armed", {}) + weak_verdicts = trade_analysis.get("weak", {}) approval_queue = query_approval_queue(store) if is_live else [] for row in approval_queue: # attach advisory for the human approver - adv = catalyst.get(str(row.get("symbol") or "").upper()) + sym = str(row.get("symbol") or "").upper() + adv = catalyst.get(sym) if adv: row["catalyst"] = adv + verdict = armed_verdicts.get(sym) or weak_verdicts.get(sym) + if verdict: + row["ai"] = verdict + triggers = query_armed_triggers(store) if is_live else {"triggers": [], "armed": 0} + for trow in triggers.get("triggers", []): # attach AI verdict per armed name + sym = str(trow.get("symbol") or "").upper() + verdict = armed_verdicts.get(sym) or weak_verdicts.get(sym) + if verdict: + trow["ai"] = verdict return { "generated_at": datetime.now(timezone.utc).isoformat(), "execution_mode": self.execution_mode, @@ -117,9 +133,10 @@ class DashboardState: "available_dates": _available_dates(store), "pnl": query_session_pnl(store, for_date=for_date), "risk": query_risk_state(store, for_date=for_date), - "triggers": query_armed_triggers(store) if is_live else {"triggers": [], "armed": 0}, + "triggers": triggers, "equity_curve": query_equity_curve(store, for_date=for_date), "catalyst": catalyst, + "trade_analysis": trade_analysis, "approval_queue": approval_queue, "ready_signals": query_ready_signals_snapshot(store) if is_live else [], # watch_states is a LIVE concept (and the heaviest projection diff --git a/api/static/dashboard.html b/api/static/dashboard.html index 548ea7c..8cf8a1b 100644 --- a/api/static/dashboard.html +++ b/api/static/dashboard.html @@ -220,6 +220,14 @@ .st-weak .trig-state { color: #7d8590; } .st-weak .trig-fill { background: #39414d; } .st-waiting .trig-state { color: #7d8590; } .st-waiting .trig-fill { background: #30363d; } .st-armed.near .trig-bar { box-shadow: 0 0 0 1px #f1c40f55; } + /* AI trade-analysis verdict chip + EOD note */ + .ai-chip { display: inline-block; padding: 0 5px; margin-right: 4px; border-radius: 3px; + font: 700 9px var(--mono); text-transform: uppercase; letter-spacing: .3px; vertical-align: middle; } + .ai-note { margin: 0 0 9px; padding: 8px 11px; background: #6e40c911; border: 1px solid #6e40c944; + border-radius: 6px; font: 500 11px var(--sans); color: var(--text); } + .ai-note .ai-note-h { font: 700 9px var(--mono); text-transform: uppercase; letter-spacing: .5px; + color: #a371f7; margin-right: 6px; } + .ai-note .ai-note-take { color: var(--muted); font-size: 10px; } /* intraday equity sparkline */ .pnl-spark { flex: 1.5 1 200px; } .eqspark { margin-top: 2px; } .eqspark svg { display: block; } @@ -292,6 +300,7 @@

Armed Triggers 0live proximity to the opening-range breakout · closest to fire on top

+
No armed gappers — waiting for the open / first bars.
@@ -541,6 +550,29 @@ function renderWatch(ws) { // Armed-trigger proximity radar — for each loaded gapper, a bar showing where the // LIVE price sits between the stop (left) and the breakout trigger (the marker). // The fill grows toward the marker as price nears the break; crossing it = fired. +// Local-LLM AI verdict chip. ai = {decision, confidence, summary, concerns}. +const AI_COLORS = { pursue: '#2ea043', avoid: '#f85149', monitor: '#d29922', defer: '#8b949e' }; +function aiChip(ai) { + if (!ai || !ai.decision || ai.decision === 'none') return ''; + const c = AI_COLORS[ai.decision] || '#8b949e'; + const conf = ai.confidence != null ? ' ' + Math.round(ai.confidence * 100) + '%' : ''; + const tip = esc((ai.summary || '') + + (ai.concerns && ai.concerns.length ? ' — ' + ai.concerns.join('; ') : '')); + return '🤖 ' + esc(ai.decision) + conf + ''; +} + +function renderEodNote(ta) { + const host = $("ai-eod-note"); + if (!host) return; + const eod = ta && ta.eod && ta.eod[""]; + if (!eod || !eod.summary) { host.style.display = 'none'; host.innerHTML = ''; return; } + const takeaways = (eod.concerns && eod.concerns.length) + ? '
' + eod.concerns.map(esc).join(' · ') + '
' : ''; + host.innerHTML = '🤖 AI session note' + esc(eod.summary) + takeaways; + host.style.display = ''; +} + const TRIG_ORDER = { fired: 0, filled: 1, armed: 2, weak: 3, waiting: 4 }; function renderTriggers(tw) { const arr = (tw && tw.triggers) || []; @@ -571,7 +603,7 @@ function renderTriggers(tw) { + (hasBar ? '
' : '') + '' + '
' + distTxt + '
' - + '
' + (t.catalyst ? esc(t.catalyst) : '') + '
' + + '
' + aiChip(t.ai) + (t.catalyst ? esc(t.catalyst) : '') + '
' + ''; }).join(''); } @@ -659,6 +691,7 @@ function render(d) { fr.style.color = (rk.data_age_s == null) ? "" : (rk.data_age_s > 180 ? "#ff4d4f" : rk.data_age_s > 90 ? "#f1c40f" : ""); renderTriggers(d.triggers); + renderEodNote(d.trade_analysis); renderEquity(d.equity_curve); const accts = d.accounts || []; diff --git a/config.py b/config.py index 0a9c664..78cb6b6 100644 --- a/config.py +++ b/config.py @@ -169,6 +169,12 @@ class OllamaConfig(BaseModel): # Phase 2 score blend (ships OFF; separate from enrichment so turning on the # advisory/dashboard does NOT silently change trade scoring). catalyst_score_enabled: bool = False + # AI trade analysis (advisory layer): armed "pursue?" verdict, weak-setup + # diagnosis, closed-trade post-mortems, EOD session narrative. OFF the hot + # path, advisory only — never gates a trade. Ships OFF. + trade_analysis_enabled: bool = False + trade_analysis_interval_seconds: int = 180 + trade_analysis_batch_limit: int = 8 # max setups analyzed per pass (GPU budget) @classmethod def from_env(cls, env: dict[str, str] | None = None) -> "OllamaConfig": @@ -209,6 +215,9 @@ class OllamaConfig(BaseModel): dilution_veto_enabled=b("NEWS_DILUTION_VETO_ENABLED", False), dilution_veto_min_conviction=f("NEWS_DILUTION_VETO_CONVICTION", 0.6), catalyst_score_enabled=b("NEWS_CATALYST_SCORE_ENABLED", False), + trade_analysis_enabled=b("TRADE_ANALYSIS_ENABLED", False), + trade_analysis_interval_seconds=i("TRADE_ANALYSIS_INTERVAL_SECONDS", 180), + trade_analysis_batch_limit=i("TRADE_ANALYSIS_BATCH_LIMIT", 8), ) diff --git a/run_live_paper.py b/run_live_paper.py index 672c26e..967699d 100644 --- a/run_live_paper.py +++ b/run_live_paper.py @@ -50,6 +50,11 @@ from research.ingestion.scheduler import Scheduler from research.ingestion.watcher_task import ResearchWatchlistProvider from research.ingestion.discovery import run_discovery, screen_universe from research.ingestion.news_enrichment import enrich_recent_news +from strategy.evaluation.trade_analysis import ( + run_closed_trade_analysis, + run_session_narrative, + run_trade_analysis, +) from config import OllamaConfig from research.multi_schema import open_research_db from runtime.exit_manager import LiveExitManager @@ -840,10 +845,32 @@ def main(argv: list[str] | None = None) -> int: return (f"enriched={res['enriched']} skipped={res['skipped']} " f"errors={res['errors']}") + # AI trade analysis — advisory only, OFF the hot path. Reads the in-memory + # armed-trigger book + the day's closed trades and writes the local LLM's read + # (pursue/avoid verdict, weak-setup diagnosis, post-mortems, EOD note) into + # ai_trade_analysis_cache for the dashboard. Gated by ollama.enabled AND the + # trade_analysis flag; a context_hash skips unchanged setups so a stable board + # doesn't keep hitting the GPU. Degrades to a no-op if Ollama is down. + def step_analyze(): + if not (ollama_cfg.enabled and ollama_cfg.trade_analysis_enabled): + return "skipped (disabled)" + from storage.projections import query_session_pnl + sess = _now_session_date() + snap = rt["book"].snapshot() + trades = (query_session_pnl(rt["store"], session_id=rt["session_id"]) or {}).get("trades", []) + a = run_trade_analysis(rt["research_con"], snap, ollama_cfg, sess) + p = run_closed_trade_analysis(rt["research_con"], trades, ollama_cfg, sess) + e = run_session_narrative(rt["research_con"], snap, trades, ollama_cfg, sess) + return (f"setups={a['analyzed']} postmortem={p['analyzed']} " + f"eod={e['analyzed']} (skipped={a['skipped'] + p['skipped'] + e['skipped']})") + scheduler = Scheduler() scheduler.add("enrich", step_enrich, float(ollama_cfg.enrichment_interval_seconds), enabled=ollama_cfg.enabled and ollama_cfg.enrichment_enabled) + scheduler.add("analyze", step_analyze, + float(ollama_cfg.trade_analysis_interval_seconds), + enabled=ollama_cfg.enabled and ollama_cfg.trade_analysis_enabled) scheduler.add("discover", step_discover, float(os.environ.get("DISCOVER_INTERVAL_SECONDS", "300")), enabled=_flag("DISCOVER_ENABLED", "1")) @@ -869,7 +896,7 @@ def main(argv: list[str] | None = None) -> int: for task in scheduler.tasks: task.last_run = -10**9 results = {} - for name in ("discover", "ingest", "enrich", "watch", "arm", "sync", "execute", "exits", "eod"): + for name in ("discover", "ingest", "enrich", "watch", "arm", "analyze", "sync", "execute", "exits", "eod"): for task in scheduler.tasks: if task.name == name and task.enabled: results.update(scheduler_run_one(task)) diff --git a/storage/pg_schema.sql b/storage/pg_schema.sql index 10b2d8e..3986013 100644 --- a/storage/pg_schema.sql +++ b/storage/pg_schema.sql @@ -412,6 +412,27 @@ CREATE INDEX IF NOT EXISTS idx_news_catalyst_cache_symbol CREATE INDEX IF NOT EXISTS idx_news_catalyst_cache_enriched ON news_catalyst_cache (enriched_at); +-- AI trade analysis (advisory): the local LLM's read of armed/weak setups, +-- closed trades, and the day's session. One current row per (type, symbol, day); +-- context_hash lets a pass skip re-calling the model when nothing changed. +-- analysis_type: 'armed' | 'weak' | 'postmortem' | 'eod'. symbol '' = session-level. +CREATE TABLE IF NOT EXISTS ai_trade_analysis_cache ( + analysis_type VARCHAR, + symbol VARCHAR, + session_date DATE, + context_hash VARCHAR, + decision VARCHAR, -- 'pursue'|'avoid'|'monitor'|'defer'|'none' + confidence DOUBLE PRECISION, + summary VARCHAR, + concerns VARCHAR, -- JSON array of strings + detail VARCHAR, -- JSON blob of the structured input context + model VARCHAR, + analyzed_at TIMESTAMP DEFAULT current_timestamp, + PRIMARY KEY (analysis_type, symbol, session_date) +); +CREATE INDEX IF NOT EXISTS idx_ai_trade_analysis_session + ON ai_trade_analysis_cache (session_date, analysis_type); + CREATE TABLE IF NOT EXISTS telemetry_events ( id VARCHAR PRIMARY KEY, event_type VARCHAR, diff --git a/storage/projections.py b/storage/projections.py index 085b6b1..55889ea 100644 --- a/storage/projections.py +++ b/storage/projections.py @@ -791,3 +791,46 @@ def query_catalyst_advisory(store, lookback_hours: int = 8) -> dict: "headline": headline, } return out + + +def query_trade_analysis(store, for_date: str | None = None) -> dict: + """Local-LLM AI trade analysis for the dashboard. + + {analysis_type: {symbol: {decision, confidence, summary, concerns}}} for a + session — 'armed'/'weak'/'postmortem' keyed by symbol, 'eod' under ''. Reads + ai_trade_analysis_cache directly; never raises, {} on any fault (table absent + / analysis never ran) so the dashboard degrades cleanly.""" + try: + if for_date and str(for_date).lower() not in ("live", "today", ""): + sess = for_date + else: + row = store.con.execute( + "SELECT MAX(session_date) FROM ai_trade_analysis_cache").fetchone() + sess = row[0] if row else None + if sess is None: + return {} + rows = store.con.execute( + "SELECT analysis_type, symbol, decision, confidence, summary, concerns " + "FROM ai_trade_analysis_cache WHERE session_date = ? " + "ORDER BY analyzed_at DESC", + [sess], + ).fetchall() + except Exception: # noqa: BLE001 + return {} + out: dict[str, dict] = {} + for atype, sym, decision, confidence, summary, concerns in rows: + bucket = out.setdefault(atype, {}) + key = (sym or "").upper() + if key in bucket: # newest first -> keep latest per symbol + continue + try: + concerns_list = json.loads(concerns) if concerns else [] + except (TypeError, ValueError): + concerns_list = [] + bucket[key] = { + "decision": decision, + "confidence": round(float(confidence), 3) if confidence is not None else None, + "summary": summary, + "concerns": concerns_list, + } + return out diff --git a/strategy/evaluation/trade_analysis.py b/strategy/evaluation/trade_analysis.py new file mode 100644 index 0000000..4f7872b --- /dev/null +++ b/strategy/evaluation/trade_analysis.py @@ -0,0 +1,394 @@ +"""Local-LLM (Ollama) AI analysis of trades — advisory layer. + +Stdlib-only. Same proven shape as ``research/ingestion/news_enrichment.py``: +ask Ollama for STRUCTURED JSON, parse tolerantly, cache in Postgres, degrade to +a no-op on any failure. Runs OFF the hot trading path (interval Scheduler) and +NEVER gates a trade — every verdict is advisory, surfaced on the dashboard. + +Four analyses (``analysis_type``): + * ``armed`` — a loaded setup: does it still make sense to pursue? (pursue/ + avoid/monitor + confidence + concerns) + * ``weak`` — a too-soft setup: why it failed and whether a real thesis remains + * ``postmortem`` — a closed trade: what went wrong, the management lesson + * ``eod`` — a once-a-day plain-English session narrative + +Design rules mirror the enrichment layer: ``?`` placeholders (translated by +storage/db_pg), bounded per pass by a batch limit, and a ``context_hash`` so an +unchanged setup is not re-sent to the GPU. +""" + +from __future__ import annotations + +import hashlib +import json +import logging +import urllib.request +from dataclasses import dataclass, field +from datetime import date + +logger = logging.getLogger(__name__) + +# Verdict vocabulary. 'none' = retrospective/narrative analyses with no call to make. +DECISIONS = {"pursue", "avoid", "monitor", "defer", "none"} + +# JSON schema handed to Ollama's ``format`` so decoding is constrained at +# generation time (one parser for all four analyses). +ANALYSIS_SCHEMA = { + "type": "object", + "properties": { + "decision": {"type": "string", "enum": sorted(DECISIONS)}, + "confidence": {"type": "number", "minimum": 0.0, "maximum": 1.0}, + "summary": {"type": "string"}, + "concerns": {"type": "array", "items": {"type": "string"}}, + }, + "required": ["decision", "confidence", "summary", "concerns"], +} + + +@dataclass +class TradeAnalysis: + """The model's read of one setup/trade/session. Validated on parse.""" + + decision: str = "none" + confidence: float = 0.0 + summary: str = "" + concerns: list[str] = field(default_factory=list) + + def as_row(self) -> dict: + return { + "decision": self.decision, + "confidence": self.confidence, + "summary": self.summary, + "concerns": self.concerns, + } + + +def _clamp(value, lo: float, hi: float, default: float) -> float: + try: + return max(lo, min(hi, float(value))) + except (TypeError, ValueError): + return default + + +def _parse(raw: str) -> TradeAnalysis | None: + """Tolerant parse of the model's JSON. None on unrecoverable failure.""" + try: + data = json.loads(raw) + except (TypeError, ValueError): + return None + if not isinstance(data, dict): + return None + decision = str(data.get("decision", "none")).strip().lower() + if decision not in DECISIONS: + decision = "none" + concerns_raw = data.get("concerns", []) or [] + if isinstance(concerns_raw, str): + concerns_raw = [concerns_raw] + concerns = [str(c).strip()[:200] for c in concerns_raw if str(c).strip()][:6] + return TradeAnalysis( + decision=decision, + confidence=_clamp(data.get("confidence", 0.0), 0.0, 1.0, 0.0), + summary=str(data.get("summary", "") or "")[:280], + concerns=concerns, + ) + + +def context_hash(payload: dict) -> str: + """Stable short hash of an analysis input, so an unchanged setup isn't re-sent.""" + blob = json.dumps(payload, sort_keys=True, default=str) + return hashlib.sha256(blob.encode()).hexdigest()[:16] + + +# --------------------------------------------------------------------------- +# Prompt builders (pure — unit tested) +# --------------------------------------------------------------------------- + +_CONTRACT = ( + "Respond with a STRICT JSON object only, no prose, with exactly these keys:\n" + ' "decision": one of [avoid, defer, monitor, none, pursue]\n' + ' "confidence": number 0.0..1.0\n' + ' "summary": one short sentence (max ~280 chars)\n' + ' "concerns": array of short strings (0-6 items)\n\n' +) + + +def build_armed_prompt(ctx: dict) -> str: + return ( + "You are a small-cap momentum day-trading coach. A gap-and-go setup is " + "ARMED (loaded to fire on an opening-range breakout). Judge whether it " + "still makes sense to PURSUE as a long, or to AVOID (e.g. dilutive/over-" + "extended/thin) or MONITOR. Weigh the gap, relative volume, range, " + "distance to trigger, and the catalyst together.\n\n" + + _CONTRACT + + "Setup:\n" + json.dumps(ctx, default=str) + ) + + +def build_weak_prompt(ctx: dict) -> str: + return ( + "You are a small-cap momentum day-trading coach. This gap setup is too " + "WEAK to fire (soft gap / low relative volume / narrow range). Explain " + "briefly WHY it's weak and whether there's still a real thesis worth " + "watching for a re-trigger. Use decision MONITOR if worth watching, " + "DEFER if not now, AVOID if it's a trap.\n\n" + + _CONTRACT + + "Setup:\n" + json.dumps(ctx, default=str) + ) + + +def build_postmortem_prompt(ctx: dict) -> str: + return ( + "You are a trading-performance coach. Review this CLOSED day-trade " + "(r_multiple = realized R, exit_reason = how it ended). In the summary " + "give the single biggest management lesson; list concerns as concrete, " + "actionable lessons. Use decision 'none' (this is retrospective).\n\n" + + _CONTRACT + + "Trade:\n" + json.dumps(ctx, default=str) + ) + + +def build_eod_prompt(ctx: dict) -> str: + return ( + "You are a trading-desk analyst. Write a brief END-OF-DAY note over the " + "day's armed names and closed trades: what worked, what didn't, and " + "patterns to watch tomorrow. Summary = the narrative; concerns = the key " + "takeaways. Use decision 'none'.\n\n" + + _CONTRACT + + "Session:\n" + json.dumps(ctx, default=str) + ) + + +# --------------------------------------------------------------------------- +# Ollama call +# --------------------------------------------------------------------------- + +def _call(prompt: str, cfg, *, use_schema: bool = True) -> TradeAnalysis | None: + """POST a prompt to Ollama and parse the structured reply. None on any failure.""" + fmt = ANALYSIS_SCHEMA if use_schema else "json" + try: + req = urllib.request.Request( + f"{cfg.host}/api/generate", + data=json.dumps( + { + "model": cfg.model, + "prompt": prompt, + "stream": False, + "format": fmt, + "options": { + "temperature": getattr(cfg, "temperature", 0.3), + "num_predict": getattr(cfg, "max_tokens", 256), + }, + } + ).encode(), + headers={"Content-Type": "application/json"}, + ) + with urllib.request.urlopen(req, timeout=getattr(cfg, "timeout_seconds", 30)) as resp: + body = json.loads(resp.read().decode()) + response = body.get("response") + except Exception: # noqa: BLE001 + logger.debug("trade analysis unavailable", exc_info=True) + return None + return _parse(response) if response else None + + +def analyze_armed_setup(ctx: dict, cfg) -> TradeAnalysis | None: + return _call(build_armed_prompt(ctx), cfg) + + +def analyze_weak_setup(ctx: dict, cfg) -> TradeAnalysis | None: + return _call(build_weak_prompt(ctx), cfg) + + +def analyze_closed_trade(ctx: dict, cfg) -> TradeAnalysis | None: + return _call(build_postmortem_prompt(ctx), cfg) + + +def analyze_session(ctx: dict, cfg) -> TradeAnalysis | None: + return _call(build_eod_prompt(ctx), cfg) + + +# --------------------------------------------------------------------------- +# Persistence (single Postgres datastore; ``?`` placeholders) +# --------------------------------------------------------------------------- + +def _cached_hash(con, analysis_type: str, symbol: str, session_date) -> str | None: + row = con.execute( + "SELECT context_hash FROM ai_trade_analysis_cache " + "WHERE analysis_type = ? AND symbol = ? AND session_date = ? LIMIT 1", + [analysis_type, symbol, session_date], + ).fetchone() + return row[0] if row else None + + +def _store_analysis(con, analysis_type, symbol, session_date, chash, analysis, detail, model): + concerns = json.dumps(analysis.concerns) + detail_json = json.dumps(detail, default=str) + exists = con.execute( + "SELECT 1 FROM ai_trade_analysis_cache " + "WHERE analysis_type = ? AND symbol = ? AND session_date = ? LIMIT 1", + [analysis_type, symbol, session_date], + ).fetchone() + if exists: + con.execute( + "UPDATE ai_trade_analysis_cache SET context_hash = ?, decision = ?, " + "confidence = ?, summary = ?, concerns = ?, detail = ?, model = ?, " + "analyzed_at = current_timestamp " + "WHERE analysis_type = ? AND symbol = ? AND session_date = ?", + [chash, analysis.decision, analysis.confidence, analysis.summary, + concerns, detail_json, model, analysis_type, symbol, session_date], + ) + return + con.execute( + "INSERT INTO ai_trade_analysis_cache (" + "analysis_type, symbol, session_date, context_hash, decision, confidence, " + "summary, concerns, detail, model) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)", + [analysis_type, symbol, session_date, chash, analysis.decision, + analysis.confidence, analysis.summary, concerns, detail_json, model], + ) + + +def _armed_context(row: dict) -> dict: + """Pluck the decision-relevant fields off an ArmedTrigger snapshot row.""" + return { + "symbol": row.get("symbol"), + "gap_pct": row.get("gap"), + "rvol": row.get("rvol"), + "trigger": row.get("trigger"), + "stop": row.get("stop"), + "range_pct": row.get("range_pct"), + "dist_to_trigger": row.get("dist"), + "catalyst": row.get("catalyst") or "", + } + + +def run_trade_analysis(con, snapshot: list[dict], cfg, session_date, limit: int | None = None) -> dict: + """Analyze armed + weak setups from a trigger-book snapshot. NEVER raises. + + Skips a setup whose context is unchanged since the last pass (context_hash), + so a stable board doesn't keep hitting the GPU. Bounded by ``limit``.""" + limit = limit if limit is not None else getattr(cfg, "trade_analysis_batch_limit", 8) + counts = {"analyzed": 0, "skipped": 0, "errors": 0} + rows = [r for r in (snapshot or []) if r.get("state") in ("armed", "weak")] + for row in rows[:limit]: + atype = row["state"] # 'armed' | 'weak' + sym = (row.get("symbol") or "").upper() + if not sym: + continue + ctx = _armed_context(row) + chash = context_hash({"t": atype, **ctx}) + try: + if _cached_hash(con, atype, sym, session_date) == chash: + counts["skipped"] += 1 + continue + except Exception as exc: # noqa: BLE001 + logger.warning("trade analysis cache read failed: %s", exc) + return counts + analysis = (analyze_armed_setup if atype == "armed" else analyze_weak_setup)(ctx, cfg) + if analysis is None: + counts["errors"] += 1 # don't poison cache; retry next pass + continue + try: + _store_analysis(con, atype, sym, session_date, chash, analysis, ctx, cfg.model) + counts["analyzed"] += 1 + except Exception as exc: # noqa: BLE001 + counts["errors"] += 1 + logger.debug("trade analysis store failed: %s", exc, exc_info=True) + return counts + + +def run_closed_trade_analysis(con, trades: list[dict], cfg, session_date, limit: int | None = None) -> dict: + """Post-mortem each closed trade (dedup by outcome). NEVER raises.""" + limit = limit if limit is not None else getattr(cfg, "trade_analysis_batch_limit", 8) + counts = {"analyzed": 0, "skipped": 0, "errors": 0} + for t in (trades or [])[:limit]: + sym = (t.get("symbol") or "").upper() + if not sym: + continue + ctx = { + "symbol": sym, + "r_multiple": t.get("r_multiple"), + "realized_pnl": t.get("realized_pnl"), + "exit_reason": t.get("exit_reason"), + } + chash = context_hash({"t": "postmortem", **ctx}) + try: + if _cached_hash(con, "postmortem", sym, session_date) == chash: + counts["skipped"] += 1 + continue + except Exception as exc: # noqa: BLE001 + logger.warning("postmortem cache read failed: %s", exc) + return counts + analysis = analyze_closed_trade(ctx, cfg) + if analysis is None: + counts["errors"] += 1 + continue + try: + _store_analysis(con, "postmortem", sym, session_date, chash, analysis, ctx, cfg.model) + counts["analyzed"] += 1 + except Exception as exc: # noqa: BLE001 + counts["errors"] += 1 + logger.debug("postmortem store failed: %s", exc, exc_info=True) + return counts + + +def run_session_narrative(con, snapshot: list[dict], trades: list[dict], cfg, session_date) -> dict: + """One end-of-day note over the day's armed names + closed trades. NEVER raises.""" + counts = {"analyzed": 0, "skipped": 0, "errors": 0} + armed = [r.get("symbol") for r in (snapshot or []) if r.get("state") in ("armed", "fired", "filled")] + ctx = { + "armed_symbols": armed, + "trades": [{"symbol": t.get("symbol"), "r_multiple": t.get("r_multiple"), + "exit_reason": t.get("exit_reason")} for t in (trades or [])], + } + chash = context_hash({"t": "eod", **ctx}) + try: + if _cached_hash(con, "eod", "", session_date) == chash: + counts["skipped"] += 1 + return counts + except Exception as exc: # noqa: BLE001 + logger.warning("eod cache read failed: %s", exc) + return counts + analysis = analyze_session(ctx, cfg) + if analysis is None: + counts["errors"] += 1 + return counts + try: + _store_analysis(con, "eod", "", session_date, chash, analysis, ctx, cfg.model) + counts["analyzed"] += 1 + except Exception as exc: # noqa: BLE001 + counts["errors"] += 1 + logger.debug("eod store failed: %s", exc, exc_info=True) + return counts + + +def trade_analysis_map(con, session_date) -> dict: + """{analysis_type: {symbol: advisory}} for a session. Never raises. + + Read model for the dashboard. ``eod`` lands under the '' symbol key.""" + try: + rows = con.execute( + "SELECT analysis_type, symbol, decision, confidence, summary, concerns " + "FROM ai_trade_analysis_cache WHERE session_date = ? " + "ORDER BY analyzed_at DESC", + [session_date], + ).fetchall() + except Exception as exc: # noqa: BLE001 + logger.warning("trade_analysis_map query failed: %s", exc) + return {} + out: dict[str, dict] = {} + for atype, sym, decision, confidence, summary, concerns in rows: + bucket = out.setdefault(atype, {}) + key = (sym or "").upper() + if key in bucket: # newest first -> keep latest per symbol + continue + try: + concerns_list = json.loads(concerns) if concerns else [] + except (TypeError, ValueError): + concerns_list = [] + bucket[key] = { + "decision": decision, + "confidence": round(float(confidence), 3) if confidence is not None else None, + "summary": summary, + "concerns": concerns_list, + } + return out diff --git a/tests/unit/storage/test_trade_analysis_projection.py b/tests/unit/storage/test_trade_analysis_projection.py new file mode 100644 index 0000000..c3157f3 --- /dev/null +++ b/tests/unit/storage/test_trade_analysis_projection.py @@ -0,0 +1,62 @@ +"""Tests for the AI trade-analysis dashboard projection.""" + +import json +from datetime import date + +import pytest + +from storage.event_store import EventStore +from storage.projections import query_trade_analysis + + +@pytest.fixture +def store(): + s = EventStore(":memory:") + yield s + s.close() + + +SESS = date(2026, 6, 29) + + +def _seed(store, atype, symbol, decision, confidence, summary, concerns, sess=SESS): + store.con.execute( + "INSERT INTO ai_trade_analysis_cache (analysis_type, symbol, session_date, " + "context_hash, decision, confidence, summary, concerns, detail, model) " + "VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)", + [atype, symbol, sess, "h", decision, confidence, summary, + json.dumps(concerns), "{}", "test-model"], + ) + + +def test_empty_when_no_rows(store): + assert query_trade_analysis(store) == {} + + +def test_groups_by_type_and_symbol(store): + _seed(store, "armed", "AAA", "pursue", 0.8, "clean gap", ["thin float"]) + _seed(store, "weak", "BBB", "monitor", 0.4, "soft", []) + _seed(store, "postmortem", "AAA", "none", 0.0, "stopped out", ["entered late"]) + _seed(store, "eod", "", "none", 0.0, "decent day", ["press winners"]) + + ta = query_trade_analysis(store) + assert ta["armed"]["AAA"]["decision"] == "pursue" + assert ta["armed"]["AAA"]["concerns"] == ["thin float"] + assert ta["weak"]["BBB"]["decision"] == "monitor" + assert ta["postmortem"]["AAA"]["summary"] == "stopped out" + # the EOD note lands under the '' symbol key + assert ta["eod"][""]["summary"] == "decent day" + + +def test_scopes_to_latest_session(store): + _seed(store, "armed", "OLD", "avoid", 0.5, "yesterday", [], sess=date(2026, 6, 26)) + _seed(store, "armed", "NEW", "pursue", 0.9, "today", [], sess=SESS) + ta = query_trade_analysis(store) # MAX(session_date) -> only the latest day + assert "NEW" in ta["armed"] and "OLD" not in ta["armed"] + + +def test_for_date_selects_past_session(store): + _seed(store, "armed", "OLD", "avoid", 0.5, "yesterday", [], sess=date(2026, 6, 26)) + _seed(store, "armed", "NEW", "pursue", 0.9, "today", [], sess=SESS) + ta = query_trade_analysis(store, for_date="2026-06-26") + assert "OLD" in ta["armed"] and "NEW" not in ta["armed"] diff --git a/tests/unit/strategy/test_trade_analysis.py b/tests/unit/strategy/test_trade_analysis.py new file mode 100644 index 0000000..7fdcab4 --- /dev/null +++ b/tests/unit/strategy/test_trade_analysis.py @@ -0,0 +1,211 @@ +"""AI trade-analysis (Ollama) tests. + +The LLM call is mocked at ``urllib.request.urlopen`` (no network); the +persistence + dedupe tests run against an in-memory Postgres schema. Same +graceful-degradation contract as the news enrichment layer: any failure -> None, +never raises. +""" + +import json +from datetime import date + +import pytest + +from strategy.evaluation import trade_analysis as ta +from storage.db import get_connection + + +# -------------------------------------------------------------------------- +# parsing +# -------------------------------------------------------------------------- + +class _FakeResp: + def __init__(self, body: bytes): + self._body = body + + def read(self): + return self._body + + def __enter__(self): + return self + + def __exit__(self, *a): + return False + + +def _reply(payload: dict) -> _FakeResp: + return _FakeResp(json.dumps({"response": json.dumps(payload)}).encode()) + + +def test_parse_happy_path(): + a = ta._parse(json.dumps({ + "decision": "pursue", "confidence": 0.8, + "summary": "Clean FDA gap with volume", "concerns": ["thin float"], + })) + assert a.decision == "pursue" and a.confidence == 0.8 + assert a.summary == "Clean FDA gap with volume" + assert a.concerns == ["thin float"] + + +def test_parse_validates_enum_and_clamps(): + a = ta._parse(json.dumps({"decision": "bogus", "confidence": 5, "summary": "x"})) + assert a.decision == "none" # unknown -> none + assert a.confidence == 1.0 # clamped to [0,1] + + +def test_parse_concerns_string_coerced_to_list(): + a = ta._parse(json.dumps({"decision": "avoid", "confidence": 0.2, + "summary": "dilutive", "concerns": "offering risk"})) + assert a.concerns == ["offering risk"] + + +def test_parse_malformed_returns_none(): + assert ta._parse("not json") is None + assert ta._parse(json.dumps([1, 2, 3])) is None + + +def test_context_hash_stable_and_sensitive(): + h1 = ta.context_hash({"a": 1, "b": 2}) + h2 = ta.context_hash({"b": 2, "a": 1}) # key order irrelevant + h3 = ta.context_hash({"a": 1, "b": 3}) + assert h1 == h2 and h1 != h3 + + +# -------------------------------------------------------------------------- +# prompt builders (pure) +# -------------------------------------------------------------------------- + +def test_prompt_builders_include_context_and_contract(): + ctx = {"symbol": "ABCD", "gap_pct": 22.0, "rvol": 8.0} + for build in (ta.build_armed_prompt, ta.build_weak_prompt, + ta.build_postmortem_prompt, ta.build_eod_prompt): + p = build(ctx) + assert "ABCD" in p + assert '"decision"' in p and '"concerns"' in p # the JSON contract + + +def test_analyze_armed_setup_over_mock(monkeypatch): + monkeypatch.setattr(ta.urllib.request, "urlopen", + lambda *a, **k: _reply({ + "decision": "pursue", "confidence": 0.7, + "summary": "ok", "concerns": [], + })) + a = ta.analyze_armed_setup({"symbol": "ABCD", "gap_pct": 20.0}, _Cfg()) + assert a is not None and a.decision == "pursue" + + +def test_call_network_down_returns_none(monkeypatch): + def boom(*a, **k): + raise OSError("refused") + monkeypatch.setattr(ta.urllib.request, "urlopen", boom) + assert ta.analyze_armed_setup({"symbol": "X"}, _Cfg()) is None + + +# -------------------------------------------------------------------------- +# persistence + dedupe (in-memory Postgres) +# -------------------------------------------------------------------------- + +class _Cfg: + host = "http://localhost:11434" + model = "test-model" + timeout_seconds = 5 + temperature = 0.3 + max_tokens = 256 + trade_analysis_batch_limit = 8 + + +@pytest.fixture +def con(): + c = get_connection(":memory:") + yield c + c.close() + + +SESS = date(2026, 6, 29) + + +def test_run_trade_analysis_persists_and_dedupes(con, monkeypatch): + snap = [ + {"symbol": "AAA", "state": "armed", "gap": 20.0, "rvol": 6.0, + "trigger": 5.1, "stop": 4.8, "range_pct": 0.05, "dist": -0.01, "catalyst": "FDA nod"}, + {"symbol": "BBB", "state": "weak", "gap": 4.0, "rvol": 2.1, + "trigger": 2.0, "stop": 1.95, "range_pct": 0.01, "dist": -0.2, "catalyst": ""}, + {"symbol": "CCC", "state": "waiting"}, # not analyzed + ] + monkeypatch.setattr(ta, "analyze_armed_setup", + lambda ctx, cfg: ta.TradeAnalysis("pursue", 0.8, "go", ["thin"])) + monkeypatch.setattr(ta, "analyze_weak_setup", + lambda ctx, cfg: ta.TradeAnalysis("monitor", 0.4, "soft", [])) + + res = ta.run_trade_analysis(con, snap, _Cfg(), SESS) + assert res["analyzed"] == 2 and res["errors"] == 0 + + rows = con.execute( + "SELECT analysis_type, symbol, decision FROM ai_trade_analysis_cache " + "ORDER BY symbol").fetchall() + assert rows == [("armed", "AAA", "pursue"), ("weak", "BBB", "monitor")] + + # second pass, unchanged snapshot -> all skipped (context_hash) + res2 = ta.run_trade_analysis(con, snap, _Cfg(), SESS) + assert res2["analyzed"] == 0 and res2["skipped"] == 2 + + +def test_run_trade_analysis_skips_when_llm_unavailable(con, monkeypatch): + snap = [{"symbol": "AAA", "state": "armed", "gap": 20.0, "rvol": 6.0}] + monkeypatch.setattr(ta, "analyze_armed_setup", lambda ctx, cfg: None) + res = ta.run_trade_analysis(con, snap, _Cfg(), SESS) + assert res["analyzed"] == 0 and res["errors"] == 1 + # cache not poisoned + assert con.execute("SELECT count(*) FROM ai_trade_analysis_cache").fetchone()[0] == 0 + + +def test_run_closed_trade_analysis_persists(con, monkeypatch): + trades = [{"symbol": "AAA", "r_multiple": -1.0, "realized_pnl": -120.0, + "exit_reason": "stop_loss"}] + monkeypatch.setattr(ta, "analyze_closed_trade", + lambda ctx, cfg: ta.TradeAnalysis("none", 0.0, "stopped out", + ["entered extended"])) + res = ta.run_closed_trade_analysis(con, trades, _Cfg(), SESS) + assert res["analyzed"] == 1 + row = con.execute( + "SELECT analysis_type, symbol, summary FROM ai_trade_analysis_cache").fetchone() + assert row[0] == "postmortem" and row[1] == "AAA" + + +def test_run_session_narrative_writes_single_eod_row(con, monkeypatch): + snap = [{"symbol": "AAA", "state": "armed"}] + trades = [{"symbol": "AAA", "r_multiple": 1.5, "exit_reason": "profit_target"}] + monkeypatch.setattr(ta, "analyze_session", + lambda ctx, cfg: ta.TradeAnalysis("none", 0.0, "solid day", ["press winners"])) + res = ta.run_session_narrative(con, snap, trades, _Cfg(), SESS) + assert res["analyzed"] == 1 + row = con.execute( + "SELECT analysis_type, symbol, summary FROM ai_trade_analysis_cache " + "WHERE analysis_type = 'eod'").fetchone() + assert row[1] == "" and row[2] == "solid day" + # unchanged -> skipped on a second pass + assert ta.run_session_narrative(con, snap, trades, _Cfg(), SESS)["skipped"] == 1 + + +def test_config_from_env_reads_trade_analysis_flags(): + from config import OllamaConfig + cfg = OllamaConfig.from_env({ + "TRADE_ANALYSIS_ENABLED": "1", + "TRADE_ANALYSIS_INTERVAL_SECONDS": "90", + "TRADE_ANALYSIS_BATCH_LIMIT": "4", + }) + assert cfg.trade_analysis_enabled is True + assert cfg.trade_analysis_interval_seconds == 90 + assert cfg.trade_analysis_batch_limit == 4 + # defaults off + assert OllamaConfig.from_env({}).trade_analysis_enabled is False + + +def test_trade_analysis_map_groups_by_type(con, monkeypatch): + snap = [{"symbol": "AAA", "state": "armed", "gap": 20.0}] + monkeypatch.setattr(ta, "analyze_armed_setup", + lambda ctx, cfg: ta.TradeAnalysis("avoid", 0.9, "dilutive", ["offering"])) + ta.run_trade_analysis(con, snap, _Cfg(), SESS) + m = ta.trade_analysis_map(con, SESS) + assert "armed" in m and m["armed"]["AAA"]["decision"] == "avoid" + assert m["armed"]["AAA"]["concerns"] == ["offering"]