Add AI trade analysis (advisory): armed/weak verdict, post-mortems, EOD note

Extends the local-LLM layer from news to TRADES, reusing the proven enrichment
pattern (structured JSON, Postgres cache, off-hot-path scheduler, graceful
degradation). Advisory only — never gates a trade.

strategy/evaluation/trade_analysis.py: four analyses keyed by analysis_type —
  * armed      — does a loaded setup still make sense to pursue? (pursue/avoid/
                 monitor + confidence + concerns), weighing numbers + catalyst
  * weak       — why a too-soft setup failed and whether a thesis remains
  * postmortem — closed-trade lesson (uses r_multiple / exit_reason)
  * eod        — once-a-day session narrative
Schema-constrained decoding (ANALYSIS_SCHEMA), tolerant _parse, and a
context_hash so an unchanged board isn't re-sent to the GPU. run_* helpers
upsert into ai_trade_analysis_cache; trade_analysis_map / query_trade_analysis
read it back. NEVER raise.

Wiring:
- ai_trade_analysis_cache table (storage/pg_schema.sql)
- OllamaConfig.trade_analysis_{enabled,interval_seconds,batch_limit} + env
- run_live_paper step_analyze on the Scheduler, gated by ollama.enabled AND
  TRADE_ANALYSIS_ENABLED (ships OFF); reads the in-memory trigger book + day P&L
- query_trade_analysis projection; api snapshots() attaches the verdict to armed
  triggers + approval rows and exposes trade_analysis
- dashboard: per-name AI verdict chip (pursue=green/avoid=red/monitor=amber) +
  an "AI session note" panel
- .env.example documents the toggles

Tests: parse/enum/clamp, prompt builders, persistence + context_hash dedupe,
closed-trade + EOD paths, config from_env, and the dashboard projection. 230 pass.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01M5HTMkaadGDnAPtXe2hWwg

.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)

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

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 @@
 <main>
   <section class="panel full">
     <h2>Armed Triggers <span id="trig-count" class="count">0</span><span class="hint">live proximity to the opening-range breakout &middot; closest to fire on top</span></h2>
+    <div id="ai-eod-note" class="ai-note" style="display:none"></div>
     <div id="armed-triggers"><span class="empty">No armed gappers — waiting for the open / first bars.</span></div>
   </section>
 
@@ -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 '<span class="ai-chip" title="' + tip + '" style="background:' + c +
+    '22;color:' + c + ';border:1px solid ' + c + '55">🤖 ' + esc(ai.decision) + conf + '</span>';
+}
+
+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)
+    ? '<div class="ai-note-take">' + eod.concerns.map(esc).join(' &middot; ') + '</div>' : '';
+  host.innerHTML = '<span class="ai-note-h">🤖 AI session note</span>' + 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 ? '<div class="trig-fill" style="width:' + fill.toFixed(1) + '%"></div><div class="trig-trigline" style="left:' + trigPos.toFixed(1) + '%"></div>' : '')
       + '</div>'
       + '<div class="trig-dist ' + distCls + '">' + distTxt + '</div>'
-      + '<div class="trig-cat">' + (t.catalyst ? esc(t.catalyst) : '') + '</div>'
+      + '<div class="trig-cat">' + aiChip(t.ai) + (t.catalyst ? esc(t.catalyst) : '') + '</div>'
       + '</div>';
   }).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 || [];

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),
         )
 
 

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))

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,

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

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

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"]

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"]