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 || [];

catalyst_eval.py

@@ -0,0 +1,41 @@
+#!/usr/bin/env python3
+"""A/B a local Ollama model on catalyst-labeling accuracy against a fixed set.
+
+Usage:
+    PYTHONPATH=. python catalyst_eval.py                       # default model
+    PYTHONPATH=. python catalyst_eval.py --model qwen3:14b     # A/B a bigger model
+    PYTHONPATH=. python catalyst_eval.py --no-schema           # plain JSON format
+
+Runs the hand-labeled fixture in ``research/ingestion/catalyst_eval.py`` through
+the live model and prints type accuracy, is_dilutive precision/recall (the veto's
+basis), and the type mistakes. Pick the model that wins here for OLLAMA_MODEL.
+"""
+
+import argparse
+
+from research.ingestion.catalyst_eval import format_report, run_eval
+
+
+def main(argv=None) -> int:
+    p = argparse.ArgumentParser(description=__doc__)
+    p.add_argument("--model", default="qwen2.5:7b-instruct")
+    p.add_argument("--host", default="http://localhost:11434")
+    p.add_argument("--timeout", type=int, default=30)
+    p.add_argument("--no-schema", action="store_true",
+                   help="send plain 'format: json' instead of the JSON schema")
+    args = p.parse_args(argv)
+
+    metrics, _ = run_eval(
+        host=args.host, model=args.model, timeout=args.timeout,
+        use_schema=not args.no_schema,
+    )
+    print(format_report(metrics, model=args.model))
+    if metrics["scored"] == 0:
+        print(f"\nNo predictions scored — is Ollama running at {args.host} "
+              f"with model '{args.model}' pulled?")
+        return 1
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

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

research/ingestion/catalyst_eval.py

@@ -0,0 +1,165 @@
+"""Label-quality evaluation for the Ollama catalyst classifier.
+
+Runs a fixed, hand-labeled set of representative small-cap headlines through
+``classify_headline`` and scores the model's labels against the ground truth.
+This is the objective basis for (a) confirming the few-shot / schema prompt
+helps and (b) A/B-ing models (``qwen3:14b`` vs ``qwen2.5:7b-instruct``) on the
+SAME inputs.
+
+The scoring (`score_predictions`) is PURE — no Ollama, no DB — so it is unit
+tested directly. ``run_eval`` is the thin live wrapper that calls the model.
+
+The single most decision-relevant metric is **is_dilutive precision/recall**:
+that flag is what the Phase-2 dilution veto acts on, so a false negative lets a
+dilutive trap through and a false positive vetoes a good name.
+"""
+
+from __future__ import annotations
+
+from research.ingestion.news_enrichment import CatalystAnalysis, classify_headline
+
+# A representative, hand-labeled fixture spanning the catalyst vocabulary.
+# ``sentiment_sign`` is the EXPECTED sign for the stock: +1 bullish, -1 bearish,
+# 0 neutral/ambiguous. It is scored with a deadband so near-zero reads pass.
+FIXTURE: list[dict] = [
+    {"headline": "Acme Bio Announces $30 Million Registered Direct Offering Priced At-The-Market",
+     "catalyst_type": "offering_dilution", "is_dilutive": True, "sentiment_sign": -1},
+    {"headline": "NanoTech Prices Underwritten Public Offering of 8 Million Shares at $2.10",
+     "catalyst_type": "offering_dilution", "is_dilutive": True, "sentiment_sign": -1},
+    {"headline": "QuantumX Files Form S-3 Shelf Registration for Up to $200 Million",
+     "catalyst_type": "offering_dilution", "is_dilutive": True, "sentiment_sign": -1},
+    {"headline": "CureGen Receives FDA Approval for Its Lead Oncology Therapy",
+     "catalyst_type": "fda_approval", "is_dilutive": False, "sentiment_sign": 1},
+    {"headline": "TrialCo Reports Positive Topline Phase 3 Results Meeting Primary Endpoint",
+     "catalyst_type": "clinical_trial", "is_dilutive": False, "sentiment_sign": 1},
+    {"headline": "BioX Drug Fails to Meet Primary Endpoint in Phase 3 Study",
+     "catalyst_type": "clinical_trial", "is_dilutive": False, "sentiment_sign": -1},
+    {"headline": "RetailCo Reports Q4 Revenue Up 35%, Beating Analyst Estimates",
+     "catalyst_type": "earnings", "is_dilutive": False, "sentiment_sign": 1},
+    {"headline": "WidgetCorp Misses Q2 Earnings as Margins Compress",
+     "catalyst_type": "earnings", "is_dilutive": False, "sentiment_sign": -1},
+    {"headline": "TargetCo to Be Acquired by MegaCorp for $12.00 Per Share in Cash",
+     "catalyst_type": "ma_acquisition", "is_dilutive": False, "sentiment_sign": 1},
+    {"headline": "TechCo Signs Multi-Year Supply Agreement With a Major Automaker",
+     "catalyst_type": "partnership_contract", "is_dilutive": False, "sentiment_sign": 1},
+    {"headline": "SoftCo Raises Full-Year Revenue Guidance Above Consensus",
+     "catalyst_type": "guidance_update", "is_dilutive": False, "sentiment_sign": 1},
+    {"headline": "PharmaCo Receives FDA Complete Response Letter for Its Drug Application",
+     "catalyst_type": "regulatory", "is_dilutive": False, "sentiment_sign": -1},
+    {"headline": "MicroCap Announces 1-for-20 Reverse Stock Split to Regain Listing Compliance",
+     "catalyst_type": "stock_split_reverse", "is_dilutive": False, "sentiment_sign": -1},
+    {"headline": "GadgetCo Upgraded to Buy at a Major Bank, Price Target Raised",
+     "catalyst_type": "analyst_rating", "is_dilutive": False, "sentiment_sign": 1},
+    {"headline": "BoardCo Authorizes $100 Million Share Repurchase Program",
+     "catalyst_type": "insider_buyback", "is_dilutive": False, "sentiment_sign": 1},
+    {"headline": "CEO of HoldingCo to Present at an Upcoming Investor Conference",
+     "catalyst_type": "none", "is_dilutive": False, "sentiment_sign": 0},
+]
+
+
+def _sign(value, deadband: float = 0.15) -> int:
+    """-1 / 0 / +1 with a neutral deadband so near-zero reads count as neutral."""
+    if value is None:
+        return 0
+    if value > deadband:
+        return 1
+    if value < -deadband:
+        return -1
+    return 0
+
+
+def _field(pred, name):
+    """Read a field off a CatalystAnalysis or a plain dict (keeps scoring pure)."""
+    if pred is None:
+        return None
+    if isinstance(pred, dict):
+        return pred.get(name)
+    return getattr(pred, name, None)
+
+
+def score_predictions(cases: list[tuple]) -> dict:
+    """Score (expected_dict, predicted) pairs. PURE — no Ollama/DB.
+
+    ``predicted`` may be a ``CatalystAnalysis``, a dict, or ``None`` (model
+    failed → counted as an error, not scored). Returns accuracy for the
+    catalyst type and sentiment sign, plus precision/recall/F1 for the
+    is_dilutive flag (the veto's basis) and the list of type mistakes."""
+    n = len(cases)
+    type_hits = sent_hits = errors = scored = 0
+    tp = fp = fn = tn = 0
+    confusion: list[dict] = []
+    for exp, pred in cases:
+        if pred is None:
+            errors += 1
+            continue
+        scored += 1
+        p_type = _field(pred, "catalyst_type")
+        if p_type == exp["catalyst_type"]:
+            type_hits += 1
+        else:
+            confusion.append({
+                "headline": exp.get("headline", ""),
+                "expected": exp["catalyst_type"], "predicted": p_type,
+            })
+        e_dil, p_dil = bool(exp["is_dilutive"]), bool(_field(pred, "is_dilutive"))
+        if p_dil and e_dil:
+            tp += 1
+        elif p_dil and not e_dil:
+            fp += 1
+        elif not p_dil and e_dil:
+            fn += 1
+        else:
+            tn += 1
+        if "sentiment_sign" in exp and _sign(_field(pred, "sentiment")) == exp["sentiment_sign"]:
+            sent_hits += 1
+
+    def _ratio(a, b):
+        return round(a / b, 4) if b else None
+
+    precision = _ratio(tp, tp + fp)
+    recall = _ratio(tp, tp + fn)
+    f1 = (round(2 * precision * recall / (precision + recall), 4)
+          if precision and recall else None)
+    return {
+        "n": n, "scored": scored, "errors": errors,
+        "type_accuracy": _ratio(type_hits, scored),
+        "sentiment_sign_accuracy": _ratio(sent_hits, scored),
+        "dilutive": {"tp": tp, "fp": fp, "fn": fn, "tn": tn,
+                     "precision": precision, "recall": recall, "f1": f1},
+        "confusion": confusion,
+    }
+
+
+def run_eval(cases: list[dict] | None = None, **kwargs) -> tuple[dict, list[tuple]]:
+    """Classify each fixture headline live, then score. Returns (metrics, pairs).
+
+    ``kwargs`` pass straight through to ``classify_headline`` (host, model,
+    timeout, use_schema, ...). A model failure on a case yields ``None`` for
+    that prediction (counted as an error)."""
+    cases = cases if cases is not None else FIXTURE
+    pairs = [
+        (c, classify_headline(c["headline"], c.get("snippet", ""),
+                              c.get("tickers", ""), **kwargs))
+        for c in cases
+    ]
+    return score_predictions(pairs), pairs
+
+
+def format_report(metrics: dict, model: str = "") -> str:
+    """Human-readable metrics block for the CLI."""
+    d = metrics["dilutive"]
+    def pct(x):
+        return f"{x * 100:5.1f}%" if isinstance(x, (int, float)) else "  n/a"
+    lines = [
+        f"Catalyst label eval  —  model={model or '?'}",
+        f"  cases: {metrics['n']}   scored: {metrics['scored']}   errors: {metrics['errors']}",
+        f"  catalyst_type accuracy : {pct(metrics['type_accuracy'])}",
+        f"  sentiment-sign accuracy: {pct(metrics['sentiment_sign_accuracy'])}",
+        f"  is_dilutive  P={pct(d['precision'])} R={pct(d['recall'])} F1={pct(d['f1'])}"
+        f"  (tp={d['tp']} fp={d['fp']} fn={d['fn']} tn={d['tn']})",
+    ]
+    if metrics["confusion"]:
+        lines.append("  type misses:")
+        for m in metrics["confusion"]:
+            lines.append(f"    [{m['expected']} -> {m['predicted']}] {m['headline'][:70]}")
+    return "\n".join(lines)

research/ingestion/news_enrichment.py

@@ -41,6 +41,24 @@ CATALYST_TYPES = {
     "none",
 }
 
+# JSON schema handed to Ollama's ``format`` field so the model is constrained at
+# decode time (enum-locked catalyst_type, typed numbers/bool) — far fewer malformed
+# rows than free-form ``"format": "json"``. Built from CATALYST_TYPES so the two
+# never drift. Current Ollama enforces this grammar during generation.
+CATALYST_SCHEMA = {
+    "type": "object",
+    "properties": {
+        "catalyst_type": {"type": "string", "enum": sorted(CATALYST_TYPES)},
+        "sentiment": {"type": "number", "minimum": -1.0, "maximum": 1.0},
+        "conviction": {"type": "number", "minimum": 0.0, "maximum": 1.0},
+        "is_dilutive": {"type": "boolean"},
+        "rationale": {"type": "string"},
+    },
+    "required": [
+        "catalyst_type", "sentiment", "conviction", "is_dilutive", "rationale",
+    ],
+}
+
 # Cheap stdlib floor for the dilution veto so it still works when Ollama is down.
 # The LLM CONFIRMS dilution; this regex is the safety net, not the primary signal.
 _DILUTIVE_RE = re.compile(
@@ -95,6 +113,20 @@ def build_prompt(headline: str, snippet: str = "", tickers: str = "") -> str:
         '  "is_dilutive": true ONLY for confirmed share dilution (offering, ATM, '
         "registered direct, warrant, shelf takedown), else false\n"
         '  "rationale": one short sentence (max ~200 chars)\n\n'
+        "Examples:\n"
+        'Headline: XYZ Therapeutics Announces $50 Million Registered Direct Offering Priced At-The-Market\n'
+        '{"catalyst_type": "offering_dilution", "sentiment": -0.8, "conviction": 0.9, '
+        '"is_dilutive": true, "rationale": "Registered direct ATM offering dilutes '
+        'shareholders; bearish for the stock."}\n'
+        'Headline: ABC Pharma Receives FDA Approval for Lead Drug Candidate\n'
+        '{"catalyst_type": "fda_approval", "sentiment": 0.9, "conviction": 0.9, '
+        '"is_dilutive": false, "rationale": "FDA approval is a major de-risking '
+        'catalyst; strongly bullish."}\n'
+        'Headline: QRS Inc. Reports Q3 Revenue Up 40%, Beats Analyst Estimates\n'
+        '{"catalyst_type": "earnings", "sentiment": 0.7, "conviction": 0.75, '
+        '"is_dilutive": false, "rationale": "Earnings beat with strong revenue '
+        'growth; bullish."}\n\n'
+        "Now classify this news:\n"
         + ctx
     )
 
@@ -143,11 +175,16 @@ def classify_headline(
     timeout: int = 30,
     temperature: float = 0.3,
     max_tokens: int = 256,
+    use_schema: bool = True,
 ) -> CatalystAnalysis | None:
-    """Ask a local Ollama model to classify a headline. None on any failure."""
+    """Ask a local Ollama model to classify a headline. None on any failure.
+
+    ``use_schema`` (default) constrains decoding with ``CATALYST_SCHEMA``; set it
+    False to fall back to plain ``"format": "json"`` (e.g. older Ollama)."""
     if not (headline or snippet):
         return None
     prompt = build_prompt(headline, snippet, tickers)
+    fmt = CATALYST_SCHEMA if use_schema else "json"
     try:
         req = urllib.request.Request(
             f"{host}/api/generate",
@@ -156,7 +193,7 @@ def classify_headline(
                     "model": model,
                     "prompt": prompt,
                     "stream": False,
-                    "format": "json",
+                    "format": fmt,
                     "options": {
                         "temperature": temperature,
                         "num_predict": max_tokens,

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/research/test_catalyst_eval.py

@@ -0,0 +1,97 @@
+"""Tests for the catalyst label-quality eval.
+
+``score_predictions`` is pure (no Ollama/DB); ``run_eval`` is exercised over a
+mocked classifier so the harness logic is covered without a live model.
+"""
+
+from research.ingestion import catalyst_eval as ce
+from research.ingestion.news_enrichment import CatalystAnalysis
+
+
+def _exp(ctype, dil, sign, headline="h"):
+    return {"headline": headline, "catalyst_type": ctype,
+            "is_dilutive": dil, "sentiment_sign": sign}
+
+
+def test_score_predictions_perfect():
+    cases = [
+        (_exp("fda_approval", False, 1), CatalystAnalysis("fda_approval", 0.9, 0.9, False)),
+        (_exp("offering_dilution", True, -1),
+         CatalystAnalysis("offering_dilution", -0.8, 0.9, True)),
+    ]
+    m = ce.score_predictions(cases)
+    assert m["scored"] == 2 and m["errors"] == 0
+    assert m["type_accuracy"] == 1.0
+    assert m["sentiment_sign_accuracy"] == 1.0
+    assert m["dilutive"]["precision"] == 1.0 and m["dilutive"]["recall"] == 1.0
+    assert m["confusion"] == []
+
+
+def test_score_predictions_counts_none_as_error_not_scored():
+    cases = [
+        (_exp("earnings", False, 1), None),
+        (_exp("earnings", False, 1), CatalystAnalysis("earnings", 0.7, 0.7, False)),
+    ]
+    m = ce.score_predictions(cases)
+    assert m["n"] == 2 and m["scored"] == 1 and m["errors"] == 1
+    assert m["type_accuracy"] == 1.0  # only the scored one counts
+
+
+def test_score_predictions_dilutive_precision_recall():
+    cases = [
+        # false negative: real dilution the model missed
+        (_exp("offering_dilution", True, -1), CatalystAnalysis("other", 0.0, 0.3, False)),
+        # false positive: model flagged dilution that isn't
+        (_exp("earnings", False, 1), CatalystAnalysis("earnings", 0.5, 0.6, True)),
+        # true positive
+        (_exp("offering_dilution", True, -1),
+         CatalystAnalysis("offering_dilution", -0.7, 0.8, True)),
+        # true negative
+        (_exp("fda_approval", False, 1), CatalystAnalysis("fda_approval", 0.9, 0.9, False)),
+    ]
+    d = ce.score_predictions(cases)["dilutive"]
+    assert (d["tp"], d["fp"], d["fn"], d["tn"]) == (1, 1, 1, 1)
+    assert d["precision"] == 0.5 and d["recall"] == 0.5
+
+
+def test_score_predictions_records_type_misses():
+    cases = [(_exp("guidance_update", False, 1, "Raises guidance"),
+              CatalystAnalysis("earnings", 0.4, 0.5, False))]
+    m = ce.score_predictions(cases)
+    assert m["type_accuracy"] == 0.0
+    assert m["confusion"][0]["expected"] == "guidance_update"
+    assert m["confusion"][0]["predicted"] == "earnings"
+
+
+def test_sentiment_sign_deadband():
+    # predicted sentiment within the deadband counts as neutral
+    cases = [(_exp("none", False, 0), CatalystAnalysis("none", 0.05, 0.1, False))]
+    assert ce.score_predictions(cases)["sentiment_sign_accuracy"] == 1.0
+
+
+def test_run_eval_over_mocked_classifier(monkeypatch):
+    # a "perfect" model that echoes each fixture's expected labels
+    def fake_classify(headline, snippet="", tickers="", **kwargs):
+        for c in ce.FIXTURE:
+            if c["headline"] == headline:
+                sign = c["sentiment_sign"]
+                return CatalystAnalysis(
+                    catalyst_type=c["catalyst_type"],
+                    sentiment=0.8 * sign, conviction=0.8,
+                    is_dilutive=c["is_dilutive"],
+                )
+        return None
+
+    monkeypatch.setattr(ce, "classify_headline", fake_classify)
+    metrics, pairs = ce.run_eval()
+    assert len(pairs) == len(ce.FIXTURE)
+    assert metrics["errors"] == 0
+    assert metrics["type_accuracy"] == 1.0
+    assert metrics["dilutive"]["precision"] == 1.0
+    assert metrics["dilutive"]["recall"] == 1.0
+
+
+def test_format_report_runs():
+    cases = [(_exp("earnings", False, 1), CatalystAnalysis("earnings", 0.6, 0.6, False))]
+    out = ce.format_report(ce.score_predictions(cases), model="m")
+    assert "catalyst_type accuracy" in out and "is_dilutive" in out

tests/unit/research/test_catalyst_validation.py

@@ -0,0 +1,79 @@
+"""Tests for the catalyst→outcome validation harness (pure aggregation only)."""
+
+import validate_catalysts as vc
+
+
+def _row(otc, oth, has_cat, dil=None, score=None):
+    return {"open_to_close_pct": otc, "open_to_high_pct": oth,
+            "has_catalyst": has_cat, "is_dilutive": dil, "catalyst_score": score}
+
+
+def test_outcome_correlation_counts_and_coverage():
+    rows = [
+        _row(-5.0, 2.0, True, dil=True, score=0.1),
+        _row(8.0, 12.0, True, dil=False, score=0.8),
+        _row(1.0, 3.0, False),  # no catalyst → excluded from splits
+    ]
+    s = vc.outcome_correlation(rows)
+    assert s["total_gappers"] == 3
+    assert s["with_catalyst"] == 2
+    assert s["dilutive"]["n"] == 1
+    assert s["non_dilutive"]["n"] == 1
+
+
+def test_outcome_correlation_dilution_split_means():
+    rows = [
+        _row(-4.0, 1.0, True, dil=True),
+        _row(-6.0, 1.0, True, dil=True),
+        _row(10.0, 15.0, True, dil=False),
+    ]
+    s = vc.outcome_correlation(rows)
+    assert s["dilutive"]["mean_open_to_close_pct"] == -5.0  # (-4 + -6)/2
+    assert s["non_dilutive"]["mean_open_to_close_pct"] == 10.0
+
+
+def test_outcome_correlation_score_buckets():
+    rows = [
+        _row(0.0, 1.0, True, dil=False, score=0.1),   # low
+        _row(0.0, 1.0, True, dil=False, score=0.5),   # mid
+        _row(0.0, 1.0, True, dil=False, score=0.9),   # high
+        _row(0.0, 1.0, True, dil=False, score=None),  # none
+    ]
+    buckets = vc.outcome_correlation(rows)["by_score_bucket"]
+    assert set(buckets) == {"low", "mid", "high", "none"}
+    assert all(b["n"] == 1 for b in buckets.values())
+
+
+def test_outcome_correlation_empty_when_no_catalysts():
+    rows = [_row(3.0, 5.0, False), _row(-2.0, 1.0, False)]
+    s = vc.outcome_correlation(rows)
+    assert s["with_catalyst"] == 0
+    assert s["dilutive"]["n"] == 0 and s["non_dilutive"]["n"] == 0
+    # report renders the "run Phase 1 live first" guidance, no crash
+    assert "run" in vc._format_report(s, n_sessions=2).lower()
+
+
+def test_session_outcome_basic():
+    import pandas as pd
+    bars = pd.DataFrame({
+        "open": [10.0, 11.0, 12.0],
+        "high": [10.5, 13.0, 12.5],
+        "low": [9.8, 10.9, 11.5],
+        "close": [10.4, 12.0, 11.0],
+    })
+    out = vc._session_outcome(bars)
+    assert out["open_to_close_pct"] == 10.0   # (11.0-10.0)/10.0*100
+    assert out["open_to_high_pct"] == 30.0    # (13.0-10.0)/10.0*100
+
+
+def test_session_outcome_empty():
+    import pandas as pd
+    out = vc._session_outcome(pd.DataFrame())
+    assert out == {"open_to_close_pct": None, "open_to_high_pct": None}
+
+
+def test_bucket_thresholds():
+    assert vc._bucket(None) == "none"
+    assert vc._bucket(0.2) == "low"
+    assert vc._bucket(0.5) == "mid"
+    assert vc._bucket(0.9) == "high"

tests/unit/research/test_news_enrichment.py

@@ -52,6 +52,42 @@ def test_classify_headline_happy_path(monkeypatch):
     assert a.is_dilutive is False and a.rationale == "FDA nod"
 
 
+def test_classify_headline_sends_schema_by_default(monkeypatch):
+    captured = {}
+
+    def capture(req, *a, **k):
+        captured["body"] = json.loads(req.data.decode())
+        return _ollama_reply({
+            "catalyst_type": "earnings", "sentiment": 0.5,
+            "conviction": 0.6, "is_dilutive": False, "rationale": "beat",
+        })
+
+    monkeypatch.setattr(ne.urllib.request, "urlopen", capture)
+    ne.classify_headline("QRS beats estimates", tickers="QRS")
+    fmt = captured["body"]["format"]
+    # schema-constrained decoding: format is the JSON schema, enum-locked
+    assert isinstance(fmt, dict)
+    assert fmt["properties"]["catalyst_type"]["enum"] == sorted(ne.CATALYST_TYPES)
+    assert set(fmt["required"]) == {
+        "catalyst_type", "sentiment", "conviction", "is_dilutive", "rationale",
+    }
+
+
+def test_classify_headline_use_schema_false_sends_plain_json(monkeypatch):
+    captured = {}
+
+    def capture(req, *a, **k):
+        captured["body"] = json.loads(req.data.decode())
+        return _ollama_reply({
+            "catalyst_type": "other", "sentiment": 0.0,
+            "conviction": 0.0, "is_dilutive": False, "rationale": "",
+        })
+
+    monkeypatch.setattr(ne.urllib.request, "urlopen", capture)
+    ne.classify_headline("anything", use_schema=False)
+    assert captured["body"]["format"] == "json"
+
+
 def test_classify_headline_network_down_returns_none(monkeypatch):
     def boom(*a, **k):
         raise OSError("connection refused")

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

validate_catalysts.py

@@ -0,0 +1,188 @@
+#!/usr/bin/env python3
+"""Validate catalyst labels against realized gapper outcomes — BEFORE automating.
+
+For every stored gapper session that ALSO has an enriched news catalyst, this
+joins the LLM's label (catalyst_type / sentiment / conviction / is_dilutive) to
+the day's realized move (open->close % and open->high %) and reports whether the
+labels actually separate winners from losers:
+
+  * do is_dilutive-flagged names fade (lower open->close) vs the rest?
+  * do higher bullish catalyst_score names run further?
+
+Run this before flipping NEWS_DILUTION_VETO_ENABLED / NEWS_CATALYST_SCORE_ENABLED.
+
+    PYTHONPATH=. python validate_catalysts.py
+    PYTHONPATH=. python validate_catalysts.py --out /tmp/catalyst_validation.csv
+
+IMPORTANT — coverage constraint: bars are retained indefinitely, but news is only
+ingested LIVE (RSS feeds can't be replayed for past dates). So a session only has
+a catalyst here if the system was running live that day. The report prints how
+many sessions actually had both; if that's ~0, let Phase 1 advisory run live for a
+stretch first to accumulate news, then re-run.
+"""
+
+from __future__ import annotations
+
+import argparse
+import csv
+from datetime import datetime, time, timedelta
+
+from research.ingestion.news_enrichment import catalyst_score
+from research.multi_schema import open_research_db
+from learn_params import load_sessions
+
+
+def _session_outcome(bars) -> dict:
+    """open->close % and open->high % from a session's (RTH) minute bars."""
+    if bars is None or len(bars) == 0:
+        return {"open_to_close_pct": None, "open_to_high_pct": None}
+    o = float(bars["open"].iloc[0])
+    if o <= 0:
+        return {"open_to_close_pct": None, "open_to_high_pct": None}
+    close = float(bars["close"].iloc[-1])
+    high = float(bars["high"].max())
+    return {
+        "open_to_close_pct": round((close - o) / o * 100.0, 3),
+        "open_to_high_pct": round((high - o) / o * 100.0, 3),
+    }
+
+
+def _catalyst_for(con, symbol: str, session_date) -> dict | None:
+    """Latest enriched catalyst for ``symbol`` within +-1 day of the session.
+
+    The +-1 day window ties an overnight/premarket PR to the morning's gap while
+    staying tight enough not to borrow a catalyst from a different week."""
+    lo = datetime.combine(session_date - timedelta(days=1), time(0, 0))
+    hi = datetime.combine(session_date + timedelta(days=1), time(0, 0))
+    try:
+        row = con.execute(
+            "SELECT catalyst_type, sentiment, conviction, is_dilutive, headline "
+            "FROM news_catalyst_cache WHERE symbol = ? "
+            "AND enriched_at >= ? AND enriched_at < ? "
+            "ORDER BY enriched_at DESC LIMIT 1",
+            [symbol, lo, hi],
+        ).fetchone()
+    except Exception:  # noqa: BLE001 — harness must never crash on a DB fault
+        return None
+    if not row:
+        return None
+    return {"catalyst_type": row[0], "sentiment": row[1], "conviction": row[2],
+            "is_dilutive": bool(row[3]), "headline": row[4]}
+
+
+def _bucket(score) -> str:
+    if score is None:
+        return "none"
+    if score < 0.33:
+        return "low"
+    if score < 0.66:
+        return "mid"
+    return "high"
+
+
+def _mean(xs):
+    return round(sum(xs) / len(xs), 3) if xs else None
+
+
+def _group_stats(rows: list[dict]) -> dict:
+    return {
+        "n": len(rows),
+        "mean_open_to_close_pct": _mean(
+            [r["open_to_close_pct"] for r in rows if r.get("open_to_close_pct") is not None]),
+        "mean_open_to_high_pct": _mean(
+            [r["open_to_high_pct"] for r in rows if r.get("open_to_high_pct") is not None]),
+    }
+
+
+def outcome_correlation(rows: list[dict]) -> dict:
+    """PURE aggregation: split gapper outcomes by dilution flag and score bucket.
+
+    ``rows`` items need: open_to_close_pct, open_to_high_pct, has_catalyst,
+    is_dilutive, catalyst_score. No DB/Ollama — unit tested directly."""
+    with_cat = [r for r in rows if r.get("has_catalyst")]
+    buckets: dict[str, list] = {}
+    for r in with_cat:
+        buckets.setdefault(_bucket(r.get("catalyst_score")), []).append(r)
+    return {
+        "total_gappers": len(rows),
+        "with_catalyst": len(with_cat),
+        "dilutive": _group_stats([r for r in with_cat if r.get("is_dilutive")]),
+        "non_dilutive": _group_stats([r for r in with_cat if not r.get("is_dilutive")]),
+        "by_score_bucket": {k: _group_stats(v) for k, v in sorted(buckets.items())},
+    }
+
+
+def _format_report(summary: dict, n_sessions: int) -> str:
+    def line(label, s):
+        c, h = s["mean_open_to_close_pct"], s["mean_open_to_high_pct"]
+        c = f"{c:+6.2f}%" if c is not None else "   n/a"
+        h = f"{h:+6.2f}%" if h is not None else "   n/a"
+        return f"    {label:<14} n={s['n']:<4} open->close {c}   open->high {h}"
+
+    out = [
+        "Catalyst → outcome validation",
+        f"  sessions with bars: {n_sessions}   gappers: {summary['total_gappers']}   "
+        f"with catalyst: {summary['with_catalyst']}",
+    ]
+    if summary["with_catalyst"] == 0:
+        out.append("")
+        out.append("  No gapper sessions had an enriched catalyst yet. News is ingested")
+        out.append("  LIVE only — let Phase 1 advisory run for a stretch, then re-run this.")
+        return "\n".join(out)
+    out.append("  Dilution split:")
+    out.append(line("dilutive", summary["dilutive"]))
+    out.append(line("non-dilutive", summary["non_dilutive"]))
+    out.append("  Bullish catalyst_score bucket:")
+    for b, s in summary["by_score_bucket"].items():
+        out.append(line(b, s))
+    return "\n".join(out)
+
+
+def gather_rows(con, limit_sessions: int | None = None) -> list[dict]:
+    sessions = load_sessions(con)
+    if limit_sessions:
+        sessions = sessions[-limit_sessions:]
+    rows = []
+    for symbol, sess, bars, _pc, _adv in sessions:
+        outcome = _session_outcome(bars)
+        cat = _catalyst_for(con, symbol, sess)
+        rows.append({
+            "symbol": symbol,
+            "session_date": str(sess),
+            **outcome,
+            "has_catalyst": cat is not None,
+            "catalyst_type": cat["catalyst_type"] if cat else None,
+            "is_dilutive": cat["is_dilutive"] if cat else None,
+            "catalyst_score": catalyst_score(cat) if cat else None,
+            "headline": cat["headline"] if cat else None,
+        })
+    return rows
+
+
+def main(argv=None) -> int:
+    p = argparse.ArgumentParser(description="Validate catalyst labels vs gapper outcomes")
+    p.add_argument("--out", default="catalyst_validation.csv", help="CSV output path")
+    p.add_argument("--limit-sessions", type=int, default=None,
+                   help="only the most recent N (symbol,date) sessions")
+    args = p.parse_args(argv)
+
+    con = open_research_db("market")
+    rows = gather_rows(con, limit_sessions=args.limit_sessions)
+    n_sessions = len({r["session_date"] for r in rows})
+    summary = outcome_correlation(rows)
+    print(_format_report(summary, n_sessions))
+
+    if rows:
+        cols = ["symbol", "session_date", "open_to_close_pct", "open_to_high_pct",
+                "has_catalyst", "catalyst_type", "is_dilutive", "catalyst_score", "headline"]
+        with open(args.out, "w", newline="") as f:
+            w = csv.DictWriter(f, fieldnames=cols)
+            w.writeheader()
+            for r in rows:
+                w.writerow({k: r.get(k) for k in cols})
+        print(f"\n  wrote {len(rows)} rows -> {args.out}")
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())