Merge PR #2 (Ollama news/catalyst enrichment) onto master + fix review blockers

Integrates the local-LLM enrichment ADDITIVELY on top of all current work, with
every Ollama feature OFF by default — with the flags off the system behaves
exactly as before (199 unit tests pass; a full --once pass runs clean with no
Ollama). Fixes the three issues the adversarial review flagged:

- BLOCKER: keep the Alpaca News ingestion. The PR predated it; the run_live_paper
  conflict is resolved to keep BOTH step_news (fetch -> raw_news_items) AND the
  enrichment. They're complementary: ingestion feeds enrichment authoritative-
  ticker headlines instead of thin RSS.
- HIGH-1: enrichment runs on its OWN daemon thread with its OWN DB connection,
  not the scheduler thread — a slow/hung Ollama can no longer starve exits /
  guard / eod-flatten, and it no longer shares the non-thread-safe research_con.
- HIGH-2: per-headline writes are all-or-nothing — on any ticker store failure
  the partial rows are deleted so the headline fully retries (no stranded
  tickers) and enriched is counted only on full success (no false count).

Phase-2 automation (catalyst score blend, dilution veto, arming-rank multiplier)
remains gated OFF pending the medium fixes (bearish-catalyst boost, prompt-
injection hardening, veto cross-thread read) noted in the review.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

.env.example

@@ -46,6 +46,29 @@ ALPACA_NEWS_ENABLED=1
 ALPACA_NEWS_INTERVAL_SECONDS=120
 ALPACA_NEWS_LIMIT=50
 
+# ---- Local LLM news/catalyst enrichment (Ollama) ----
+# A local model classifies ingested headlines (catalyst type, sentiment,
+# conviction, dilution) into news_catalyst_cache — surfaced on the dashboard
+# approval queue. Runs off the hot path; degrades to a no-op if Ollama is down.
+# OLLAMA_ENABLED=1
+# OLLAMA_HOST=http://localhost:11434
+# OLLAMA_MODEL=qwen2.5:7b-instruct   # fits 12GB; best small-model JSON.
+#                                    # alts: llama3.1:8b-instruct-q4_K_M, mistral:7b-instruct
+# NEWS_ENRICH_ENABLED=1
+# NEWS_ENRICH_INTERVAL_SECONDS=120
+# NEWS_ENRICH_LOOKBACK_HOURS=12
+# NEWS_ENRICH_BATCH_LIMIT=50         # max headlines classified per pass (GPU budget)
+# Phase 2 (automation — each ships OFF; separate from the advisory above so the
+# dashboard never silently changes scoring):
+#   score blend — fold the catalyst into the setup quality score AND graded arming
+#   rank (bullish catalyst boosts, confirmed dilution sinks below the armed cut).
+# NEWS_CATALYST_SCORE_ENABLED=0
+# NEWS_RANK_BOOST_GAIN=2.0           # arming-rank multiplier = 1 + gain*score
+# NEWS_DILUTIVE_PENALTY=0.1          # arming-rank multiplier for a confirmed dilutive name
+#   dilution veto — block confirmed dilutive offerings before sizing.
+# NEWS_DILUTION_VETO_ENABLED=0
+# NEWS_DILUTION_VETO_CONVICTION=0.6
+
 # ============================================================
 # Live paper-trading orchestrator (run_live_paper.py)
 # ============================================================

api/main.py

@@ -35,6 +35,7 @@ from storage.projections import (
     query_account_positions_snapshot,
     query_account_summary_snapshot,
     query_approval_queue,
+    query_catalyst_advisory,
     query_fills_feed,
     query_armed_triggers,
     query_equity_curve,
@@ -100,6 +101,14 @@ class DashboardState:
         with self._lock:
             store = self.open_store()
             try:
+                # 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 {}
+                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())
+                    if adv:
+                        row["catalyst"] = adv
                 return {
                     "generated_at": datetime.now(timezone.utc).isoformat(),
                     "execution_mode": self.execution_mode,
@@ -110,7 +119,8 @@ class DashboardState:
                     "risk": query_risk_state(store, for_date=for_date),
                     "triggers": query_armed_triggers(store) if is_live else {"triggers": [], "armed": 0},
                     "equity_curve": query_equity_curve(store, for_date=for_date),
-                    "approval_queue": query_approval_queue(store) if is_live else [],
+                    "catalyst": catalyst,
+                    "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
                     # — ~99k criteria rows for a whole past day); skip it for a

api/static/dashboard.html

@@ -510,6 +510,19 @@ function riskChip(req) {
   return '<span class="risk">risk <b>' + money(dollarRisk) + '</b> &middot; ' + qty + 'sh &middot; notional <b>' + money(notional) + '</b> &middot; stop <b>' + (perShare/entry*100).toFixed(1) + '%</b></span>';
 }
 
+// Local-LLM catalyst advisory cell for the approval queue. adv = {catalyst_type,
+// sentiment, conviction, is_dilutive, rationale}. Empty when not enriched.
+function catalystCell(adv) {
+  if (!adv || !adv.catalyst_type) return '<span class="muted">—</span>';
+  const s = adv.sentiment;
+  const tone = s == null ? "" : (s > 0.15 ? "long" : (s < -0.15 ? "short" : ""));
+  const type = String(adv.catalyst_type).replace(/_/g, " ");
+  const dil = adv.is_dilutive ? ' <span class="pill short" title="confirmed dilution — Phase 2 veto candidate">⚠ DILUTIVE</span>' : '';
+  const chip = '<span class="pill ' + tone + '" title="' + esc(adv.rationale || "") + '">📰 ' + esc(type)
+    + (s != null ? ' ' + (s > 0 ? '+' : '') + Number(s).toFixed(2) : '') + '</span>';
+  return chip + dil;
+}
+
 function table(headers, rows) {
   if (!rows.length) return null;
   return "<table><thead><tr>" + headers.map(h => "<th>" + h + "</th>").join("") + "</tr></thead><tbody>" + rows.join("") + "</tbody></table>";
@@ -704,12 +717,13 @@ function render(d) {
   const aq = d.approval_queue || [];
   $("aq-count").textContent = aq.length;
   fill("approval-queue", table(
-    ["Symbol", "Side", "Risk preview", "Mode", "Actions"],
+    ["Symbol", "Side", "Risk preview", "Catalyst", "Mode", "Actions"],
     aq.map(q => {
       const r = q.execution_request || {};
       return '<tr><td class="sym">' + esc(q.symbol) + '</td>' +
         '<td><span class="pill ' + (r.side === "sell" ? "short" : "long") + '">' + esc(r.side || "buy") + '</span></td>' +
         '<td>' + riskChip(r) + '</td>' +
+        '<td>' + catalystCell(q.catalyst) + '</td>' +
         '<td class="num">' + esc(q.approval_mode || "manual") + '</td>' +
         '<td><div class="actions">' +
         '<button class="btn-approve" onclick="approve(\'' + esc(q.order_id) + '\')">Approve</button>' +

config.py

@@ -1,5 +1,6 @@
 """Configuration loading from YAML files."""
 
+import os
 from pathlib import Path
 from typing import Any
 
@@ -144,16 +145,71 @@ class NewsSourcesConfig(BaseModel):
 
 
 class OllamaConfig(BaseModel):
-    """Ollama local LLM settings for enrichment."""
+    """Ollama local LLM settings for news/catalyst enrichment."""
 
     model_config = ConfigDict(extra="ignore")
 
     enabled: bool = False
     host: str = "http://localhost:11434"
-    model: str = "mistral"
+    # qwen2.5:7b-instruct (Q4, ~4.7GB) fits a 12GB GPU with headroom and has the
+    # best small-model strict-JSON adherence. Alts: llama3.1:8b-instruct-q4_K_M,
+    # mistral:7b-instruct, qwen2.5:3b-instruct (throughput).
+    model: str = "qwen2.5:7b-instruct"
     timeout_seconds: int = 30
     max_tokens: int = 256
     temperature: float = 0.3
+    # News/catalyst enrichment (advisory layer).
+    enrichment_enabled: bool = False
+    enrichment_interval_seconds: int = 120
+    enrichment_lookback_hours: int = 12
+    enrichment_batch_limit: int = 50  # max headlines per pass (GPU budget)
+    # Phase 2 dilution veto (ships OFF).
+    dilution_veto_enabled: bool = False
+    dilution_veto_min_conviction: float = 0.6
+    # 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
+
+    @classmethod
+    def from_env(cls, env: dict[str, str] | None = None) -> "OllamaConfig":
+        """Build from OLLAMA_* / NEWS_* env vars (run_live_paper reads env, not YAML)."""
+        v = dict(os.environ)
+        if env:
+            v.update(env)
+
+        def b(key: str, default: bool) -> bool:
+            raw = v.get(key)
+            if raw is None:
+                return default
+            return raw.strip().lower() in {"1", "true", "yes", "on"}
+
+        def i(key: str, default: int) -> int:
+            try:
+                return int(v.get(key, default))
+            except (TypeError, ValueError):
+                return default
+
+        def f(key: str, default: float) -> float:
+            try:
+                return float(v.get(key, default))
+            except (TypeError, ValueError):
+                return default
+
+        return cls(
+            enabled=b("OLLAMA_ENABLED", False),
+            host=v.get("OLLAMA_HOST", cls.model_fields["host"].default),
+            model=v.get("OLLAMA_MODEL", cls.model_fields["model"].default),
+            timeout_seconds=i("OLLAMA_TIMEOUT_SECONDS", 30),
+            max_tokens=i("OLLAMA_MAX_TOKENS", 256),
+            temperature=f("OLLAMA_TEMPERATURE", 0.3),
+            enrichment_enabled=b("NEWS_ENRICH_ENABLED", False),
+            enrichment_interval_seconds=i("NEWS_ENRICH_INTERVAL_SECONDS", 120),
+            enrichment_lookback_hours=i("NEWS_ENRICH_LOOKBACK_HOURS", 12),
+            enrichment_batch_limit=i("NEWS_ENRICH_BATCH_LIMIT", 50),
+            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),
+        )
 
 
 class CapabilityRegistryConfig(BaseModel):

research/ingestion/news_enrichment.py

@@ -0,0 +1,335 @@
+"""Local-LLM (Ollama) news/catalyst enrichment — advisory layer.
+
+Stdlib-only. Mirrors the proven ``strategy/evaluation/llm_integration.py`` urllib
+call but asks Ollama for STRUCTURED JSON (``"format": "json"``) so each headline
+yields a parseable catalyst classification.
+
+Design rules (see plan):
+  * Runs OFF the hot trading path (called from the interval Scheduler).
+  * Every public function degrades gracefully — returns ``None`` / skips on any
+    failure (network down, bad JSON, missing keys). Trading is never blocked.
+  * Single Postgres datastore: SQL uses DuckDB-style ``?`` placeholders, which
+    ``storage/db_pg`` translates to psycopg2 ``%s`` (exactly like rss.py).
+"""
+
+from __future__ import annotations
+
+import json
+import logging
+import re
+import urllib.request
+from dataclasses import dataclass
+from datetime import datetime, timedelta
+
+logger = logging.getLogger(__name__)
+
+# Closed vocabulary the model must choose from. Keep in sync with build_prompt.
+CATALYST_TYPES = {
+    "offering_dilution",
+    "earnings",
+    "fda_approval",
+    "clinical_trial",
+    "ma_acquisition",
+    "partnership_contract",
+    "guidance_update",
+    "regulatory",
+    "halt_resumption",
+    "stock_split_reverse",
+    "insider_buyback",
+    "analyst_rating",
+    "other",
+    "none",
+}
+
+# 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(
+    r"\b(offering|registered\s+direct|at[-\s]the[-\s]market|atm\s+(?:facility|offering)|"
+    r"warrant|dilut\w*|shelf\s+(?:registration|offering)|priced\s+(?:public\s+)?offering|"
+    r"pricing\s+of\s+.{0,30}offering)\b|\bform\s+s-?3\b",
+    re.IGNORECASE,
+)
+
+
+def looks_dilutive(text: str) -> bool:
+    """Keyword floor for the dilution signal (used when the LLM is unavailable)."""
+    return bool(_DILUTIVE_RE.search(text or ""))
+
+
+@dataclass
+class CatalystAnalysis:
+    """A model's read of one headline. All fields validated/clamped on parse."""
+
+    catalyst_type: str = "other"
+    sentiment: float = 0.0  # [-1, 1] bearish..bullish for the stock
+    conviction: float = 0.0  # [0, 1] model confidence it's a real, tradeable catalyst
+    is_dilutive: bool = False
+    rationale: str = ""
+
+    def as_news_event_fields(self) -> dict:
+        """Map to the (previously dead) news_events columns."""
+        return {
+            "sentiment": self.sentiment,
+            "category": self.catalyst_type,
+            "is_offering": self.is_dilutive,
+            "is_earnings": self.catalyst_type == "earnings",
+            "is_halt_related": self.catalyst_type == "halt_resumption",
+        }
+
+
+def build_prompt(headline: str, snippet: str = "", tickers: str = "") -> str:
+    """Instruction prompt pinning the enum and demanding a strict JSON object."""
+    types = ", ".join(sorted(CATALYST_TYPES))
+    ctx = f"Headline: {headline}"
+    if snippet:
+        ctx += f"\nDetail: {snippet}"
+    if tickers:
+        ctx += f"\nTickers: {tickers}"
+    return (
+        "You are a small-cap trading analyst. Classify the market catalyst in this "
+        "news for a momentum day-trader. Respond with a STRICT JSON object only, no "
+        "prose, with exactly these keys:\n"
+        '  "catalyst_type": one of [' + types + "]\n"
+        '  "sentiment": number from -1.0 (very bearish) to 1.0 (very bullish) for the stock\n'
+        '  "conviction": number from 0.0 to 1.0 — how confident this is a real, tradeable catalyst\n'
+        '  "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'
+        + ctx
+    )
+
+
+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, fallback_text: str = "") -> CatalystAnalysis | None:
+    """Tolerant parse of the model's JSON string. None on unrecoverable failure."""
+    try:
+        data = json.loads(raw)
+    except (TypeError, ValueError):
+        return None
+    if not isinstance(data, dict):
+        return None
+    ctype = str(data.get("catalyst_type", "other")).strip().lower()
+    if ctype not in CATALYST_TYPES:
+        ctype = "other"
+    is_dilutive = bool(data.get("is_dilutive", False)) or (
+        ctype == "offering_dilution"
+    )
+    # belt-and-suspenders: trust the keyword floor if the model missed obvious dilution
+    if not is_dilutive and looks_dilutive(fallback_text):
+        is_dilutive = True
+    rationale = str(data.get("rationale", "") or "")[:200]
+    return CatalystAnalysis(
+        catalyst_type=ctype,
+        sentiment=_clamp(data.get("sentiment", 0.0), -1.0, 1.0, 0.0),
+        conviction=_clamp(data.get("conviction", 0.0), 0.0, 1.0, 0.0),
+        is_dilutive=is_dilutive,
+        rationale=rationale,
+    )
+
+
+def classify_headline(
+    headline: str,
+    snippet: str = "",
+    tickers: str = "",
+    *,
+    host: str = "http://localhost:11434",
+    model: str = "qwen2.5:7b-instruct",
+    timeout: int = 30,
+    temperature: float = 0.3,
+    max_tokens: int = 256,
+) -> CatalystAnalysis | None:
+    """Ask a local Ollama model to classify a headline. None on any failure."""
+    if not (headline or snippet):
+        return None
+    prompt = build_prompt(headline, snippet, tickers)
+    try:
+        req = urllib.request.Request(
+            f"{host}/api/generate",
+            data=json.dumps(
+                {
+                    "model": model,
+                    "prompt": prompt,
+                    "stream": False,
+                    "format": "json",
+                    "options": {
+                        "temperature": temperature,
+                        "num_predict": max_tokens,
+                    },
+                }
+            ).encode(),
+            headers={"Content-Type": "application/json"},
+        )
+        with urllib.request.urlopen(req, timeout=timeout) as resp:
+            body = json.loads(resp.read().decode())
+        response = body.get("response")
+    except Exception:  # noqa: BLE001
+        logger.debug("catalyst enrichment unavailable", exc_info=True)
+        return None
+    if not response:
+        return None
+    return _parse(response, fallback_text=f"{headline} {snippet}")
+
+
+# ---------------------------------------------------------------------------
+# Persistence (single Postgres datastore; ``?`` placeholders, like rss.py)
+# ---------------------------------------------------------------------------
+
+def catalyst_score(advisory: dict | None) -> float | None:
+    """Map a catalyst advisory dict -> a 0..1 'how bullish a catalyst' score.
+
+    None when there is no advisory. A bearish / no-catalyst read scores low; a
+    high-conviction bullish catalyst scores high. Used by Phase 2 to blend into
+    the setup quality score (computed OUTSIDE the pure evaluator)."""
+    if not advisory:
+        return None
+    conviction = _clamp(advisory.get("conviction", 0.0), 0.0, 1.0, 0.0)
+    sentiment = _clamp(advisory.get("sentiment", 0.0), -1.0, 1.0, 0.0)
+    return round(conviction * (0.5 + 0.5 * max(0.0, sentiment)), 4)
+
+
+def _store_cache_row(con, headline_hash, symbol, headline, source, analysis, model):
+    exists = con.execute(
+        "SELECT 1 FROM news_catalyst_cache WHERE headline_hash = ? AND symbol = ? LIMIT 1",
+        [headline_hash, symbol],
+    ).fetchone()
+    if exists:
+        return
+    con.execute(
+        "INSERT INTO news_catalyst_cache ("
+        "headline_hash, symbol, headline, source, catalyst_type, sentiment, "
+        "conviction, is_dilutive, rationale, model"
+        ") VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
+        [
+            headline_hash, symbol, headline, source, analysis.catalyst_type,
+            analysis.sentiment, analysis.conviction, analysis.is_dilutive,
+            analysis.rationale, model,
+        ],
+    )
+
+
+def _upsert_news_event(con, headline_hash, symbol, headline, source, published_at, analysis):
+    """Populate the (previously dead) news_events row for this headline+ticker."""
+    nid = f"{headline_hash}:{symbol}"
+    f = analysis.as_news_event_fields()
+    exists = con.execute(
+        "SELECT 1 FROM news_events WHERE id = ? LIMIT 1", [nid]
+    ).fetchone()
+    if exists:
+        con.execute(
+            "UPDATE news_events SET sentiment = ?, category = ?, is_offering = ?, "
+            "is_earnings = ?, is_halt_related = ? WHERE id = ?",
+            [f["sentiment"], f["category"], f["is_offering"], f["is_earnings"],
+             f["is_halt_related"], nid],
+        )
+        return
+    con.execute(
+        "INSERT INTO news_events ("
+        "id, symbol, headline, source, published_at, sentiment, category, "
+        "is_earnings, is_offering, is_halt_related"
+        ") VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
+        [nid, symbol, headline, source, published_at, f["sentiment"],
+         f["category"], f["is_earnings"], f["is_offering"], f["is_halt_related"]],
+    )
+
+
+def enrich_recent_news(con, cfg, lookback_hours: int | None = None, limit: int | None = None) -> dict:
+    """Classify recently-ingested, not-yet-enriched headlines via Ollama.
+
+    Left-anti-joins ``raw_news_items`` against ``news_catalyst_cache`` on the
+    headline hash, classifies each (one LLM call per headline, fanned out per
+    ticker), and writes the cache + populates ``news_events``. Bounded per pass
+    by ``limit`` so a backlog cannot run the GPU forever. NEVER raises."""
+    lookback_hours = lookback_hours if lookback_hours is not None else getattr(
+        cfg, "enrichment_lookback_hours", 12)
+    limit = limit if limit is not None else getattr(cfg, "enrichment_batch_limit", 50)
+    counts = {"enriched": 0, "skipped": 0, "errors": 0}
+    cutoff = datetime.utcnow() - timedelta(hours=lookback_hours)
+    try:
+        rows = con.execute(
+            "SELECT payload_hash, raw_title, raw_body_snippet, raw_tickers, source, "
+            "fetched_at FROM raw_news_items "
+            "WHERE payload_hash IS NOT NULL AND fetched_at >= ? "
+            "AND payload_hash NOT IN (SELECT DISTINCT headline_hash FROM news_catalyst_cache) "
+            "ORDER BY fetched_at DESC LIMIT ?",
+            [cutoff, limit],
+        ).fetchall()
+    except Exception as exc:  # noqa: BLE001
+        logger.warning("enrich_recent_news query failed (DB fault?): %s", exc)
+        return counts
+    for payload_hash, title, snippet, raw_tickers, source, fetched_at in rows:
+        analysis = classify_headline(
+            title or "", snippet or "", raw_tickers or "",
+            host=cfg.host, model=cfg.model, timeout=cfg.timeout_seconds,
+            temperature=cfg.temperature, max_tokens=cfg.max_tokens,
+        )
+        if analysis is None:
+            # Ollama down / unparseable — DON'T poison the cache; retry next pass.
+            counts["errors"] += 1
+            continue
+        tickers = [t.strip().upper() for t in (raw_tickers or "").split(",") if t.strip()]
+        if not tickers:
+            tickers = [""]  # sentinel: marks the headline enriched so we don't re-call
+        # All-or-nothing per headline: the connection is autocommit (no real
+        # transaction), so on any ticker failure we DELETE the partial rows for
+        # this headline. Otherwise the headline-level dedup (NOT IN cache) would
+        # exclude it forever, stranding the un-stored tickers; and we'd over-count.
+        ok = True
+        for sym in tickers:
+            try:
+                _store_cache_row(con, payload_hash, sym, title, source, analysis, cfg.model)
+                if sym:
+                    _upsert_news_event(con, payload_hash, sym, title, source, fetched_at, analysis)
+            except Exception as exc:  # noqa: BLE001
+                ok = False
+                counts["errors"] += 1
+                logger.debug("catalyst store failed for %s: %s", sym, exc, exc_info=True)
+                break
+        if ok:
+            counts["enriched"] += 1
+        else:
+            try:  # roll back the partial headline so it fully retries next pass
+                con.execute(
+                    "DELETE FROM news_catalyst_cache WHERE headline_hash = ?",
+                    [payload_hash],
+                )
+            except Exception:  # noqa: BLE001
+                pass
+            counts["skipped"] += 1
+    return counts
+
+
+def catalyst_map(con, lookback_hours: int = 8) -> dict[str, dict]:
+    """{ticker: advisory} for tickers enriched within the window (latest wins).
+
+    Mirrors ``research.ingestion.discovery.recent_news_map`` — read at arm /
+    approval time and (Phase 2) injected into the watcher. Never raises."""
+    cutoff = datetime.utcnow() - timedelta(hours=lookback_hours)
+    try:
+        rows = con.execute(
+            "SELECT symbol, catalyst_type, sentiment, conviction, is_dilutive, "
+            "rationale, headline FROM news_catalyst_cache "
+            "WHERE symbol <> '' AND enriched_at >= ? ORDER BY enriched_at DESC",
+            [cutoff],
+        ).fetchall()
+    except Exception as exc:  # noqa: BLE001
+        logger.warning("catalyst_map query failed (DB fault?): %s", exc)
+        return {}
+    out: dict[str, dict] = {}
+    for sym, ctype, sentiment, conviction, is_dilutive, rationale, headline in rows:
+        key = (sym or "").upper()
+        if key and key not in out:  # newest first -> first seen is the latest
+            out[key] = {
+                "catalyst_type": ctype,
+                "sentiment": sentiment,
+                "conviction": conviction,
+                "is_dilutive": bool(is_dilutive),
+                "rationale": rationale,
+                "headline": headline,
+            }
+    return out

run_live_paper.py

@@ -50,6 +50,8 @@ 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.alpaca_news import ingest_alpaca_news
+from research.ingestion.news_enrichment import enrich_recent_news
+from config import OllamaConfig
 from research.multi_schema import open_research_db
 from runtime.exit_manager import LiveExitManager
 from runtime.halt_guard import is_halt_gap
@@ -195,6 +197,33 @@ def build_runtime(args: argparse.Namespace) -> dict:
               f"(tuned {str(_learned.get('as_of', ''))[:10]} over {_learned.get('sessions')} "
               f"sessions; backtest pnl {_learned.get('pnl')}, entry@{_learned.get('entry_min')}min)")
 
+    ollama_cfg = OllamaConfig.from_env()
+
+    # Catalyst advisory provider, TTL-cached so a pass doesn't re-query per symbol;
+    # returns None on any fault so neither the (gated) dilution veto nor the score
+    # blend ever block/skew on a DB blip. Shared by the watcher AND the execution
+    # service (one cache). catalyst_score_provider maps the advisory -> a 0..1 score
+    # OUTSIDE the pure evaluator, and is only handed to the watcher when the score
+    # blend is explicitly enabled (so Phase 1 advisory never changes scoring).
+    _catalyst_cache: dict = {"map": {}, "at": 0.0}
+    _CATALYST_TTL = 30.0
+
+    def catalyst_provider(symbol: str):
+        import time as _time
+        from research.ingestion.news_enrichment import catalyst_map
+        now = _time.monotonic()
+        if (now - _catalyst_cache["at"]) >= _CATALYST_TTL:
+            try:
+                _catalyst_cache["map"] = catalyst_map(research_con)
+            except Exception:  # noqa: BLE001
+                _catalyst_cache["map"] = {}
+            _catalyst_cache["at"] = now
+        return _catalyst_cache["map"].get((symbol or "").upper())
+
+    def catalyst_score_provider(symbol: str):
+        from research.ingestion.news_enrichment import catalyst_score
+        return catalyst_score(catalyst_provider(symbol))
+
     watcher = Watcher(
         store,
         ResearchWatchlistProvider(
@@ -212,6 +241,9 @@ def build_runtime(args: argparse.Namespace) -> dict:
                 "WATCHER_READY_SCORE", str(_learned.get("ready_score_pct", 60.0)))),
             min_quality=float(os.environ.get("WATCHER_MIN_QUALITY", "0.30")),
         ),
+        catalyst_score_provider=(
+            catalyst_score_provider if ollama_cfg.catalyst_score_enabled else None
+        ),
     )
 
     exec_settings = ExecutionSettings.from_env()
@@ -255,6 +287,7 @@ def build_runtime(args: argparse.Namespace) -> dict:
             trading_mode=TradingModeSettings.from_env(),
             session_id=session_id,
             price_provider=latest_price,
+            catalyst_provider=catalyst_provider,
         )
         if executor
         else None
@@ -286,6 +319,8 @@ def build_runtime(args: argparse.Namespace) -> dict:
         "sync": sync,
         "book": book,
         "latest_price": latest_price,
+        "catalyst_provider": catalyst_provider,
+        "ollama_cfg": ollama_cfg,
         "session_id": session_id,
         "event_db": event_db,
     }
@@ -294,6 +329,7 @@ def build_runtime(args: argparse.Namespace) -> dict:
 def main(argv: list[str] | None = None) -> int:
     args = parse_args(argv)
     rt = build_runtime(args)
+    ollama_cfg = rt["ollama_cfg"]
     symbols: list[str] = rt["symbols"]
     client = rt["client"]
 
@@ -560,6 +596,32 @@ def main(argv: list[str] | None = None) -> int:
         news_map = recent_news_map(
             rt["research_con"], lookback_hours=int(os.environ.get("NEWS_LOOKBACK_HOURS", "8")))
         news_boost = float(os.environ.get("NEWS_RANK_BOOST", "2.0"))
+        # GATED graded ranking from the LLM catalyst read. When the score blend is
+        # off, ranking is the existing binary `news_boost if catalyst else 1.0`.
+        # When on, a bullish catalyst boosts by `1 + gain*score` and a confirmed
+        # dilutive offering is sunk below the armed cut (multiplied by a small
+        # penalty) so the book stops arming the gap-and-fade traps.
+        cat_map: dict = {}
+        if ollama_cfg.catalyst_score_enabled:
+            from research.ingestion.news_enrichment import catalyst_map as _cmap
+            try:
+                cat_map = _cmap(rt["research_con"])
+            except Exception:  # noqa: BLE001
+                cat_map = {}
+        news_rank_gain = float(os.environ.get("NEWS_RANK_BOOST_GAIN", "2.0"))
+        dilutive_penalty = float(os.environ.get("NEWS_DILUTIVE_PENALTY", "0.1"))
+
+        def _cat_mult(sym: str, has_headline: bool) -> float:
+            if not ollama_cfg.catalyst_score_enabled:
+                return news_boost if has_headline else 1.0
+            from research.ingestion.news_enrichment import catalyst_score as _cscore
+            adv = cat_map.get((sym or "").upper())
+            if not adv:
+                return 1.0
+            if adv.get("is_dilutive") and float(adv.get("conviction") or 0.0) >= (
+                    ollama_cfg.dilution_veto_min_conviction):
+                return dilutive_penalty
+            return 1.0 + news_rank_gain * (_cscore(adv) or 0.0)
         # optional MANUAL per-symbol priority nudge (SYMBOL_WEIGHTS=ABCD:1.2,XYZ:1.1)
         # — a small boost to names you want watched; keep modest so it doesn't
         # override the gap/volume logic.
@@ -611,7 +673,7 @@ def main(argv: list[str] | None = None) -> int:
                 # a fresh catalyst is WHY a small-cap runs — boost it up the rank;
                 # times an optional manual per-symbol weight (default 1.0)
                 "_score": (g.gap_pct * max(g.relative_volume, 0.1)
-                           * (news_boost if catalyst else 1.0)
+                           * _cat_mult(g.symbol, bool(catalyst))
                            * weights.get(g.symbol, 1.0)),
                 "complete": complete,
             })
@@ -784,6 +846,24 @@ def main(argv: list[str] | None = None) -> int:
             return f"+{len(res.new_items)} new / {res.item_count} fetched"
         return None
 
+    # Local-LLM (Ollama) news/catalyst enrichment — advisory layer. Runs on its
+    # OWN daemon thread with its OWN DB connection (started below), NOT the
+    # scheduler thread: a slow/hung Ollama pass (up to batch_limit x timeout) must
+    # never stall exits / guard / eod-flatten, which share the scheduler thread,
+    # and research_con is a single psycopg2 connection unsafe to share across
+    # threads. Gated by ollama.enabled AND enrichment_enabled; no-op if Ollama down.
+    def enrich_loop():
+        try:
+            econ = open_research_db("market")
+        except Exception as exc:  # noqa: BLE001
+            print(f"[enrich] no DB connection, enrichment off: {exc}", flush=True)
+            return
+        while not stop_event.wait(float(ollama_cfg.enrichment_interval_seconds)):
+            try:
+                enrich_recent_news(econ, ollama_cfg)
+            except Exception as exc:  # noqa: BLE001
+                print(f"[enrich] error: {exc}", flush=True)
+
     scheduler = Scheduler()
     scheduler.add("news", step_news, float(os.environ.get("ALPACA_NEWS_INTERVAL_SECONDS", "120")),
                   enabled=_flag("ALPACA_NEWS_ENABLED", "1"))
@@ -845,6 +925,10 @@ def main(argv: list[str] | None = None) -> int:
         trigger_thread = threading.Thread(target=trigger_loop, name="trigger", daemon=True)
         trigger_thread.start()
         print(f"[boot] fast trigger thread started (every {trigger_interval:.0f}s)")
+    # news enrichment on its OWN thread (never blocks the scheduler's trade steps)
+    if ollama_cfg.enabled and ollama_cfg.enrichment_enabled:
+        threading.Thread(target=enrich_loop, name="enrich", daemon=True).start()
+        print(f"[boot] news enrichment thread started (every {ollama_cfg.enrichment_interval_seconds}s, model {ollama_cfg.model})")
     try:
         # one immediate full pass, then interval-driven
         run_pass()

runtime/watcher.py

@@ -15,7 +15,7 @@ from __future__ import annotations
 import logging
 from dataclasses import dataclass, field
 from datetime import datetime, date
-from typing import Protocol
+from typing import Callable, Protocol
 
 import pandas as pd
 
@@ -80,10 +80,14 @@ class Watcher:
         store: EventStore,
         provider: WatchlistProvider,
         config: WatcherConfig | None = None,
+        catalyst_score_provider: Callable[[str], float | None] | None = None,
     ):
         self.store = store
         self.provider = provider
         self.config = config or WatcherConfig()
+        # optional symbol -> 0..1 catalyst score (or None). Injected so the pure
+        # evaluator stays DB-free; None provider == today's behavior exactly.
+        self.catalyst_score_provider = catalyst_score_provider
         if not self.config.session_id:
             self.config.session_id = f"session-{date.today().isoformat()}"
         # symbol -> current state for this session
@@ -181,6 +185,15 @@ class Watcher:
                 except Exception:  # noqa: BLE001 - fall back to now()
                     eval_time = None
 
+                # optional LLM catalyst score (None unless the blend is enabled
+                # AND a provider is injected) — never raises into the loop.
+                cat_score = None
+                if self.catalyst_score_provider is not None:
+                    try:
+                        cat_score = self.catalyst_score_provider(symbol)
+                    except Exception:  # noqa: BLE001
+                        cat_score = None
+
                 evaluation = evaluate_setup(
                     bars,
                     previous_close=candidate.previous_close,
@@ -188,6 +201,7 @@ class Watcher:
                     ready_score_pct=self.config.ready_score_pct,
                     evaluation_time=eval_time,
                     min_bars=self.config.min_bars,
+                    catalyst_score=cat_score,
                 )
                 result.evaluated += 1
 

storage/pg_schema.sql

@@ -390,6 +390,28 @@ CREATE INDEX IF NOT EXISTS idx_raw_news_items_fetch_attempt
 CREATE INDEX IF NOT EXISTS idx_raw_news_items_source_published
     ON raw_news_items (source, raw_published_at);
 
+-- Local-LLM (Ollama) catalyst enrichment cache. One row per (headline, ticker);
+-- headline_hash = raw_news_items.payload_hash (sha256(source|url|title)) so we
+-- dedupe and join for free. Read model for the dashboard advisory + Phase 2.
+CREATE TABLE IF NOT EXISTS news_catalyst_cache (
+    headline_hash   VARCHAR,
+    symbol          VARCHAR,
+    headline        VARCHAR,
+    source          VARCHAR,
+    catalyst_type   VARCHAR,
+    sentiment       DOUBLE PRECISION,
+    conviction      DOUBLE PRECISION,
+    is_dilutive     BOOLEAN DEFAULT FALSE,
+    rationale       VARCHAR,
+    model           VARCHAR,
+    enriched_at     TIMESTAMP DEFAULT current_timestamp,
+    PRIMARY KEY (headline_hash, symbol)
+);
+CREATE INDEX IF NOT EXISTS idx_news_catalyst_cache_symbol
+    ON news_catalyst_cache (symbol);
+CREATE INDEX IF NOT EXISTS idx_news_catalyst_cache_enriched
+    ON news_catalyst_cache (enriched_at);
+
 CREATE TABLE IF NOT EXISTS telemetry_events (
     id VARCHAR PRIMARY KEY,
     event_type VARCHAR,

storage/projections.py

@@ -758,3 +758,36 @@ def query_equity_curve(store, points: int = 200, for_date: str | None = None) ->
     if len(series) > points:  # keep the most recent N
         series = series[-points:]
     return {"equity": series, "baseline": baseline if baseline else (series[0] if series else None)}
+
+
+def query_catalyst_advisory(store, lookback_hours: int = 8) -> dict:
+    """Local-LLM news/catalyst read for the dashboard advisory column.
+
+    {ticker: {catalyst_type, sentiment, conviction, is_dilutive, rationale,
+    headline}} for tickers enriched within the window (latest per symbol). Reads
+    the news_catalyst_cache table directly (same Postgres datastore) — never
+    raises, returns {} on any fault so the dashboard degrades cleanly."""
+    from datetime import datetime, timedelta
+    cutoff = (datetime.utcnow() - timedelta(hours=lookback_hours))
+    try:
+        rows = store.con.execute(
+            "SELECT symbol, catalyst_type, sentiment, conviction, is_dilutive, "
+            "rationale, headline FROM news_catalyst_cache "
+            "WHERE symbol <> '' AND enriched_at >= ? ORDER BY enriched_at DESC",
+            [cutoff],
+        ).fetchall()
+    except Exception:  # noqa: BLE001 — table may not exist / Ollama never ran
+        return {}
+    out: dict[str, dict] = {}
+    for sym, ctype, sentiment, conviction, is_dilutive, rationale, headline in rows:
+        key = (sym or "").upper()
+        if key and key not in out:  # newest first -> first seen is latest
+            out[key] = {
+                "catalyst_type": ctype,
+                "sentiment": round(float(sentiment), 3) if sentiment is not None else None,
+                "conviction": round(float(conviction), 3) if conviction is not None else None,
+                "is_dilutive": bool(is_dilutive),
+                "rationale": rationale,
+                "headline": headline,
+            }
+    return out

strategy/evaluation/quality.py

@@ -33,11 +33,18 @@ def calculate_setup_quality(
     opening_strength: str = "neutral",
     data_quality: float = 1.0,
     thresholds: QualityThresholds | None = None,
+    catalyst_score: float | None = None,
 ) -> QualityScore:
     """Blend setup ingredients into one 0..1 quality score.
 
-    Weights: gap 20%, RVOL 25%, structure 30%, VWAP 10%,
+    Default weights (no catalyst): gap 20%, RVOL 25%, structure 30%, VWAP 10%,
     opening strength 5%, data quality 10%.
+
+    When ``catalyst_score`` (a 0..1 'how bullish a catalyst' signal, computed
+    OUTSIDE from the LLM advisory) is provided, weights re-balance to make room
+    for a 15% catalyst component: gap 15%, RVOL 20%, structure 25%, VWAP 10%,
+    opening 5%, data 10%, catalyst 15%. Passing ``None`` reproduces the exact
+    pre-catalyst score (so existing behavior/tests are unchanged).
     """
     thresholds = thresholds or QualityThresholds()
 
@@ -50,14 +57,36 @@ def calculate_setup_quality(
     )
     dq_component = min(1.0, max(0.0, data_quality))
 
-    score = (
+    components = {
-        0.20 * gap_component
+        "gap": gap_component,
-        + 0.25 * rvol_component
+        "relative_volume": rvol_component,
-        + 0.30 * structure_component
+        "structure": structure_component,
-        + 0.10 * vwap_component
+        "vwap": vwap_component,
-        + 0.05 * opening_component
+        "opening_strength": opening_component,
-        + 0.10 * dq_component
+        "data_quality": dq_component,
-    )
+    }
+
+    if catalyst_score is None:
+        score = (
+            0.20 * gap_component
+            + 0.25 * rvol_component
+            + 0.30 * structure_component
+            + 0.10 * vwap_component
+            + 0.05 * opening_component
+            + 0.10 * dq_component
+        )
+    else:
+        catalyst_component = min(1.0, max(0.0, catalyst_score))
+        components["catalyst"] = catalyst_component
+        score = (
+            0.15 * gap_component
+            + 0.20 * rvol_component
+            + 0.25 * structure_component
+            + 0.10 * vwap_component
+            + 0.05 * opening_component
+            + 0.10 * dq_component
+            + 0.15 * catalyst_component
+        )
 
     if score >= thresholds.a_grade:
         grade = "A"
@@ -71,13 +100,6 @@ def calculate_setup_quality(
     return QualityScore(
         score=round(score, 4),
         grade=grade,
-        components={
+        components=components,
-            "gap": gap_component,
-            "relative_volume": rvol_component,
-            "structure": structure_component,
-            "vwap": vwap_component,
-            "opening_strength": opening_component,
-            "data_quality": dq_component,
-        },
         tradeable=score >= thresholds.min_tradeable,
     )

strategy/evaluation/setup_evaluator.py

@@ -37,6 +37,7 @@ def evaluate_setup(
     entry_cutoff: EntryCutoffConfig | None = None,
     ready_score_pct: float = 60.0,
     min_bars: int = MIN_BARS,
+    catalyst_score: float | None = None,
 ) -> SetupEvaluationResult:
     """Evaluate a symbol's session bars for a momentum setup.
 
@@ -137,6 +138,7 @@ def evaluate_setup(
         above_vwap=above_vwap,
         opening_strength=first.opening_strength,
         data_quality=dq.score,
+        catalyst_score=catalyst_score,
     )
 
     entry_price = structure.breakout_level or price

tests/integration/test_catalyst_enrichment_flow.py

@@ -0,0 +1,83 @@
+"""End-to-end (Ollama mocked): raw headline -> enrich -> dashboard advisory.
+
+Exercises the full Phase 1 path on a single Postgres connection: seed
+raw_news_items, run the enrichment pass with the LLM call patched, then read it
+back through the dashboard projection exactly as api/main.py snapshots() does.
+"""
+
+from types import SimpleNamespace
+
+import pytest
+
+from research.ingestion import news_enrichment as ne
+from storage.db import get_connection
+from storage.projections import query_catalyst_advisory
+
+
+class _Cfg:
+    host = "http://localhost:11434"
+    model = "test-model"
+    timeout_seconds = 5
+    temperature = 0.3
+    max_tokens = 128
+    enrichment_lookback_hours = 12
+    enrichment_batch_limit = 50
+
+
+@pytest.fixture
+def con():
+    c = get_connection(":memory:")
+    yield c
+    c.close()
+
+
+def test_headline_flows_to_dashboard_advisory(con, monkeypatch):
+    # 1. a fresh dilutive-offering headline lands in the raw table (as RSS would)
+    con.execute(
+        "INSERT INTO raw_news_items (id, fetch_attempt_id, source, raw_url, raw_title, "
+        "raw_published_at, raw_body_snippet, raw_tickers, payload_hash, parser_version, "
+        "fetched_at, ingest_run_id) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, "
+        "current_timestamp, ?)",
+        ["h1", "fa1", "prnewswire", "http://x/1",
+         "SHEL Inc. announces $30M registered direct offering",
+         "", "pricing of offering", "SHEL", "h1", "rss-v2", "run1"],
+    )
+
+    # 2. Ollama replies with a structured classification (patched — no network).
+    #    urlopen is used as a context manager in classify_headline.
+    import json
+
+    class _CtxResp:
+        def __init__(self, body):
+            self._b = body
+        def read(self):
+            return self._b
+        def __enter__(self):
+            return self
+        def __exit__(self, *a):
+            return False
+    monkeypatch.setattr(
+        ne.urllib.request, "urlopen",
+        lambda *a, **k: _CtxResp(json.dumps({"response": json.dumps({
+            "catalyst_type": "offering_dilution", "sentiment": -0.8,
+            "conviction": 0.9, "is_dilutive": True,
+            "rationale": "registered direct offering dilutes holders",
+        })}).encode()),
+    )
+
+    # 3. the scheduled enrichment pass runs
+    res = ne.enrich_recent_news(con, _Cfg())
+    assert res["enriched"] == 1
+
+    # 4. the dashboard reads it via the projection (store.con == this connection)
+    store = SimpleNamespace(con=con)
+    advisory = query_catalyst_advisory(store)
+    assert "SHEL" in advisory
+    assert advisory["SHEL"]["is_dilutive"] is True
+    assert advisory["SHEL"]["catalyst_type"] == "offering_dilution"
+
+    # 5. the previously-dead news_events.sentiment column is now populated
+    sent = con.execute(
+        "SELECT sentiment FROM news_events WHERE symbol = ?", ["SHEL"]
+    ).fetchone()
+    assert sent is not None and sent[0] == -0.8

tests/unit/research/test_news_enrichment.py

@@ -0,0 +1,205 @@
+"""Local-LLM (Ollama) news/catalyst enrichment tests.
+
+The LLM call is mocked at ``urllib.request.urlopen`` (no network), matching the
+codebase's graceful-degradation contract: any failure -> ``None``, never raises.
+The persistence tests run against an in-memory Postgres schema.
+"""
+
+import io
+import json
+
+import pytest
+
+from research.ingestion import news_enrichment as ne
+from storage.db import get_connection
+
+
+# --------------------------------------------------------------------------
+# classify_headline / 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 _ollama_reply(payload: dict) -> _FakeResp:
+    """Wrap a model JSON object the way Ollama's /api/generate returns it."""
+    return _FakeResp(json.dumps({"response": json.dumps(payload)}).encode())
+
+
+def test_classify_headline_happy_path(monkeypatch):
+    monkeypatch.setattr(
+        ne.urllib.request, "urlopen",
+        lambda *a, **k: _ollama_reply({
+            "catalyst_type": "fda_approval", "sentiment": 0.9,
+            "conviction": 0.85, "is_dilutive": False, "rationale": "FDA nod",
+        }),
+    )
+    a = ne.classify_headline("BioCo gets FDA approval", tickers="BIO")
+    assert a is not None
+    assert a.catalyst_type == "fda_approval"
+    assert a.sentiment == 0.9 and a.conviction == 0.85
+    assert a.is_dilutive is False and a.rationale == "FDA nod"
+
+
+def test_classify_headline_network_down_returns_none(monkeypatch):
+    def boom(*a, **k):
+        raise OSError("connection refused")
+    monkeypatch.setattr(ne.urllib.request, "urlopen", boom)
+    assert ne.classify_headline("anything") is None
+
+
+def test_classify_headline_malformed_json_returns_none(monkeypatch):
+    monkeypatch.setattr(
+        ne.urllib.request, "urlopen",
+        lambda *a, **k: _FakeResp(json.dumps({"response": "not json at all"}).encode()),
+    )
+    assert ne.classify_headline("headline") is None
+
+
+def test_classify_headline_empty_input_returns_none():
+    assert ne.classify_headline("", "") is None
+
+
+def test_parse_clamps_and_validates():
+    a = ne._parse('{"catalyst_type":"bogus","sentiment":5,"conviction":-1}')
+    assert a.catalyst_type == "other"      # unknown enum -> other
+    assert a.sentiment == 1.0              # clamped to [-1, 1]
+    assert a.conviction == 0.0             # clamped to [0, 1]
+
+
+def test_parse_offering_type_implies_dilutive():
+    a = ne._parse('{"catalyst_type":"offering_dilution","sentiment":-0.5,"conviction":0.7,"is_dilutive":false}')
+    assert a.is_dilutive is True           # type forces the flag
+
+
+def test_parse_keyword_floor_catches_missed_dilution():
+    # model said not dilutive, but the headline text obviously is
+    a = ne._parse(
+        '{"catalyst_type":"other","sentiment":0.0,"conviction":0.3,"is_dilutive":false}',
+        fallback_text="Company announces $40M registered direct offering",
+    )
+    assert a.is_dilutive is True
+
+
+@pytest.mark.parametrize("text,expected", [
+    ("Priced public offering of 5M shares", True),
+    ("Announces at-the-market facility", True),
+    ("Files Form S-3 shelf registration", True),
+    ("Reports record Q3 earnings beat", False),
+    ("Signs partnership with Pfizer", False),
+])
+def test_looks_dilutive(text, expected):
+    assert ne.looks_dilutive(text) is expected
+
+
+def test_catalyst_score():
+    assert ne.catalyst_score(None) is None
+    assert ne.catalyst_score({"conviction": 0.0, "sentiment": 1.0}) == 0.0
+    # bearish catalyst -> low score (sentiment floored at 0 in the multiplier)
+    assert ne.catalyst_score({"conviction": 0.9, "sentiment": -0.8}) == pytest.approx(0.45)
+    assert ne.catalyst_score({"conviction": 0.9, "sentiment": 0.8}) == pytest.approx(0.81)
+
+
+# --------------------------------------------------------------------------
+# Persistence: enrich_recent_news + catalyst_map (in-memory Postgres)
+# --------------------------------------------------------------------------
+
+@pytest.fixture
+def con():
+    c = get_connection(":memory:")
+    yield c
+    c.close()
+
+
+class _Cfg:
+    """Minimal stand-in for OllamaConfig (avoids pydantic in the persist tests)."""
+    host = "http://localhost:11434"
+    model = "test-model"
+    timeout_seconds = 5
+    temperature = 0.3
+    max_tokens = 128
+    enrichment_lookback_hours = 12
+    enrichment_batch_limit = 50
+
+
+def _seed_news(con, payload_hash, title, tickers, source="prnewswire"):
+    con.execute(
+        "INSERT INTO raw_news_items (id, fetch_attempt_id, source, raw_url, raw_title, "
+        "raw_published_at, raw_body_snippet, raw_tickers, payload_hash, parser_version, "
+        "fetched_at, ingest_run_id) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, "
+        "current_timestamp, ?)",
+        [payload_hash, "fa1", source, "http://x/1", title, "", "snippet",
+         tickers, payload_hash, "rss-v2", "run1"],
+    )
+
+
+def test_enrich_recent_news_persists_and_dedupes(con, monkeypatch):
+    _seed_news(con, "hash_abc", "BioCo gets FDA approval", "BIO")
+    fixed = ne.CatalystAnalysis(
+        catalyst_type="fda_approval", sentiment=0.8, conviction=0.9,
+        is_dilutive=False, rationale="approval",
+    )
+    monkeypatch.setattr(ne, "classify_headline", lambda *a, **k: fixed)
+
+    res = ne.enrich_recent_news(con, _Cfg())
+    assert res["enriched"] == 1 and res["errors"] == 0
+
+    # cache row written
+    cache = con.execute(
+        "SELECT symbol, catalyst_type, sentiment, conviction FROM news_catalyst_cache"
+    ).fetchall()
+    assert cache == [("BIO", "fda_approval", 0.8, 0.9)]
+
+    # the previously-dead news_events.sentiment is now populated
+    ev = con.execute(
+        "SELECT symbol, sentiment, category, is_offering FROM news_events"
+    ).fetchall()
+    assert ev == [("BIO", 0.8, "fda_approval", False)]
+
+    # second pass enriches nothing (deduped by headline hash)
+    res2 = ne.enrich_recent_news(con, _Cfg())
+    assert res2["enriched"] == 0
+
+
+def test_enrich_skips_when_llm_unavailable(con, monkeypatch):
+    _seed_news(con, "hash_down", "Some headline", "XYZ")
+    monkeypatch.setattr(ne, "classify_headline", lambda *a, **k: None)
+    res = ne.enrich_recent_news(con, _Cfg())
+    assert res["enriched"] == 0 and res["errors"] == 1
+    # cache NOT poisoned — headline stays available for a retry next pass
+    assert con.execute("SELECT count(*) FROM news_catalyst_cache").fetchone()[0] == 0
+
+
+def test_enrich_fans_out_multiple_tickers(con, monkeypatch):
+    _seed_news(con, "hash_multi", "MergerCo to acquire TargetCo", "ACQ,TGT")
+    fixed = ne.CatalystAnalysis(catalyst_type="ma_acquisition", sentiment=0.6,
+                                conviction=0.8, is_dilutive=False, rationale="M&A")
+    monkeypatch.setattr(ne, "classify_headline", lambda *a, **k: fixed)
+    ne.enrich_recent_news(con, _Cfg())
+    syms = sorted(r[0] for r in con.execute(
+        "SELECT symbol FROM news_catalyst_cache").fetchall())
+    assert syms == ["ACQ", "TGT"]
+
+
+def test_catalyst_map_latest_per_symbol(con, monkeypatch):
+    _seed_news(con, "hash_dil", "BioCo announces registered direct offering", "BIO")
+    fixed = ne.CatalystAnalysis(catalyst_type="offering_dilution", sentiment=-0.7,
+                                conviction=0.85, is_dilutive=True, rationale="dilution")
+    monkeypatch.setattr(ne, "classify_headline", lambda *a, **k: fixed)
+    ne.enrich_recent_news(con, _Cfg())
+
+    m = ne.catalyst_map(con)
+    assert "BIO" in m
+    assert m["BIO"]["is_dilutive"] is True
+    assert m["BIO"]["catalyst_type"] == "offering_dilution"

tests/unit/runtime/test_watcher.py

@@ -74,6 +74,56 @@ def test_tick_emits_canonical_events(store, provider):
     assert states  # discovered -> ... -> ready transitions recorded
 
 
+def _ready_quality(store, symbol="GOOD"):
+    import json
+    events = store.query_events(event_type="signal_ready", symbol=symbol)
+    assert events, "expected a signal_ready event"
+    payload = json.loads(events[-1]["payload_json"])
+    data = payload.get("signal_data") or payload.get("payload", {}).get("signal_data")
+    return data["quality_score"]
+
+
+def test_catalyst_score_provider_lifts_quality(store, provider):
+    # baseline: no provider -> today's exact quality score
+    base = make_watcher(store, provider)
+    base.tick(SESSION_DATE)
+    baseline_q = _ready_quality(store)
+
+    # a strong bullish catalyst for GOOD raises its blended quality score
+    store2 = EventStore(":memory:")
+    try:
+        watcher = Watcher(
+            store2, provider,
+            WatcherConfig(session_id="t2", mode=EventMode.PAPER, min_quality=0.0),
+            catalyst_score_provider=lambda sym: 1.0 if sym == "GOOD" else None,
+        )
+        watcher.tick(SESSION_DATE)
+        boosted_q = _ready_quality(store2)
+    finally:
+        store2.close()
+
+    assert boosted_q > baseline_q
+
+
+def test_none_catalyst_provider_is_unchanged(store, provider):
+    # an explicit provider that returns None must reproduce the baseline exactly
+    base = make_watcher(store, provider)
+    base.tick(SESSION_DATE)
+    baseline_q = _ready_quality(store)
+
+    store2 = EventStore(":memory:")
+    try:
+        watcher = Watcher(
+            store2, provider,
+            WatcherConfig(session_id="t3", mode=EventMode.PAPER, min_quality=0.0),
+            catalyst_score_provider=lambda sym: None,
+        )
+        watcher.tick(SESSION_DATE)
+        assert _ready_quality(store2) == baseline_q
+    finally:
+        store2.close()
+
+
 def test_signal_ready_debounced_across_ticks(store, provider):
     watcher = make_watcher(store, provider)
     watcher.tick(SESSION_DATE)

tests/unit/storage/test_catalyst_advisory_projection.py

@@ -0,0 +1,44 @@
+"""Tests for the catalyst-advisory dashboard projection."""
+
+import pytest
+
+from storage.event_store import EventStore
+from storage.projections import query_catalyst_advisory
+
+
+@pytest.fixture
+def store():
+    s = EventStore(":memory:")
+    yield s
+    s.close()
+
+
+def _seed(store, headline_hash, symbol, ctype, sentiment, conviction, dilutive):
+    store.con.execute(
+        "INSERT INTO news_catalyst_cache (headline_hash, symbol, headline, source, "
+        "catalyst_type, sentiment, conviction, is_dilutive, rationale, model) "
+        "VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
+        [headline_hash, symbol, f"{symbol} news", "rss", ctype, sentiment,
+         conviction, dilutive, "why", "test-model"],
+    )
+
+
+def test_advisory_empty_when_no_rows(store):
+    assert query_catalyst_advisory(store) == {}
+
+
+def test_advisory_returns_per_symbol(store):
+    _seed(store, "h1", "BIO", "fda_approval", 0.9, 0.85, False)
+    _seed(store, "h2", "SHEL", "offering_dilution", -0.7, 0.8, True)
+
+    adv = query_catalyst_advisory(store)
+    assert adv["BIO"]["catalyst_type"] == "fda_approval"
+    assert adv["BIO"]["is_dilutive"] is False
+    assert adv["BIO"]["sentiment"] == 0.9
+    assert adv["SHEL"]["is_dilutive"] is True
+
+
+def test_advisory_skips_empty_symbol_marker(store):
+    # enrichment writes a sentinel row with symbol='' for ticker-less headlines
+    _seed(store, "h3", "", "other", 0.0, 0.2, False)
+    assert query_catalyst_advisory(store) == {}

tests/unit/strategy/test_quality_catalyst.py

@@ -0,0 +1,62 @@
+"""Phase 2: optional catalyst_score blend stays pure & backward-compatible."""
+
+import pytest
+
+from strategy.evaluation.quality import calculate_setup_quality
+
+
+_KW = dict(
+    gap_pct=0.08,
+    relative_volume=3.0,
+    structure_quality=0.7,
+    above_vwap=True,
+    opening_strength="strong",
+    data_quality=1.0,
+)
+
+
+def test_none_catalyst_reproduces_legacy_score():
+    # The exact pre-catalyst weighting (gap .20 / rvol .25 / structure .30 /
+    # vwap .10 / opening .05 / data .10) must be unchanged when score is None.
+    q = calculate_setup_quality(**_KW, catalyst_score=None)
+    expected = (
+        0.20 * 0.8 + 0.25 * 0.6 + 0.30 * 0.7 + 0.10 * 1.0
+        + 0.05 * 1.0 + 0.10 * 1.0
+    )
+    assert q.score == pytest.approx(round(expected, 4))
+    assert "catalyst" not in q.components
+
+
+def test_default_arg_matches_explicit_none():
+    assert calculate_setup_quality(**_KW).score == calculate_setup_quality(
+        **_KW, catalyst_score=None
+    ).score
+
+
+def test_strong_catalyst_raises_score_and_adds_component():
+    base = calculate_setup_quality(**_KW, catalyst_score=None).score
+    high = calculate_setup_quality(**_KW, catalyst_score=1.0)
+    assert high.score > base
+    assert high.components["catalyst"] == 1.0
+
+
+def test_zero_catalyst_lowers_score_vs_legacy():
+    # re-balancing toward a 0 catalyst pulls a strong setup down a touch
+    base = calculate_setup_quality(**_KW, catalyst_score=None).score
+    weak = calculate_setup_quality(**_KW, catalyst_score=0.0).score
+    assert weak < base
+
+
+def test_catalyst_weights_sum_to_one():
+    # an all-1.0 setup with a 1.0 catalyst must score exactly 1.0 (weights sum=1)
+    q = calculate_setup_quality(
+        gap_pct=1.0, relative_volume=10.0, structure_quality=1.0,
+        above_vwap=True, opening_strength="strong", data_quality=1.0,
+        catalyst_score=1.0,
+    )
+    assert q.score == pytest.approx(1.0)
+
+
+def test_catalyst_score_clamped():
+    q = calculate_setup_quality(**_KW, catalyst_score=5.0)
+    assert q.components["catalyst"] == 1.0

tests/unit/test_setup_evaluator.py

@@ -63,3 +63,15 @@ def test_criteria_results_cover_all_weights():
     }
     evaluated = set(result.criteria_names_passed) | set(result.criteria_names_failed)
     assert expected <= evaluated
+
+
+def test_catalyst_score_is_optional_passthrough():
+    """The Phase 2 catalyst_score is a pure pass-through: None == legacy, and a
+    strong catalyst lifts the setup's quality score without changing the gate."""
+    kw = dict(previous_close=10.0, avg_daily_volume=500_000, evaluation_time=EVAL_TIME)
+    legacy = evaluate_setup(bull_flag_bars(), **kw)
+    explicit_none = evaluate_setup(bull_flag_bars(), catalyst_score=None, **kw)
+    boosted = evaluate_setup(bull_flag_bars(), catalyst_score=1.0, **kw)
+
+    assert legacy.setups[0]["quality_score"] == explicit_none.setups[0]["quality_score"]
+    assert boosted.setups[0]["quality_score"] > legacy.setups[0]["quality_score"]

tests/unit/test_trading_execution.py

@@ -472,3 +472,70 @@ def test_repeated_backouts_bench_symbol_for_session(store):
     service._register_backout("THRASH")
     clock.advance(10_000)  # hours later, still benched
     assert service._in_cooldown("THRASH")
+
+
+# --- catalyst dilution veto (Phase 2; gated OFF by default) -----------------
+
+def _service_with_catalyst(store, advisory, *, veto_enabled, executor=None):
+    settings = ExecutionSettings(
+        enabled=True, auto_approve=False, max_orders_per_tick=2,
+        max_concurrent_positions=3, risk_per_trade_pct=0.01,
+        default_equity=100_000.0,
+        catalyst_veto_enabled=veto_enabled,
+        catalyst_veto_min_conviction=0.6,
+    )
+    return TradingExecutionService(
+        store,
+        executor=executor or FakeExecutor(),
+        settings=settings,
+        trading_mode=TradingModeSettings(execution_mode="alpaca_paper"),
+        session_id="t",
+        catalyst_provider=lambda sym: advisory,
+    )
+
+
+def test_dilution_veto_blocks_entry_and_emits_event(store):
+    emit_signal(store, symbol="DILUT")
+    advisory = {"is_dilutive": True, "conviction": 0.9,
+                "catalyst_type": "offering_dilution", "sentiment": -0.8,
+                "rationale": "registered direct offering"}
+    service = _service_with_catalyst(store, advisory, veto_enabled=True)
+
+    requested = service.request_approvals_for_ready_signals()
+    assert requested == []                       # no approval created
+    assert query_approval_queue(store) == []
+    risk_events = store.query_events(event_type="risk_rule_triggered")
+    assert any(
+        __import__("json").loads(e["payload_json"]).get("rule_type")
+        == "catalyst_dilution_veto"
+        for e in risk_events
+    )
+
+
+def test_dilution_veto_disabled_allows_entry(store):
+    emit_signal(store, symbol="DILUT")
+    advisory = {"is_dilutive": True, "conviction": 0.9}
+    service = _service_with_catalyst(store, advisory, veto_enabled=False)
+    assert len(service.request_approvals_for_ready_signals()) == 1
+
+
+def test_dilution_veto_below_conviction_allows_entry(store):
+    emit_signal(store, symbol="MAYBE")
+    advisory = {"is_dilutive": True, "conviction": 0.4}  # under the 0.6 floor
+    service = _service_with_catalyst(store, advisory, veto_enabled=True)
+    assert len(service.request_approvals_for_ready_signals()) == 1
+
+
+def test_non_dilutive_catalyst_allows_entry(store):
+    emit_signal(store, symbol="FDA")
+    advisory = {"is_dilutive": False, "conviction": 0.95,
+                "catalyst_type": "fda_approval"}
+    service = _service_with_catalyst(store, advisory, veto_enabled=True)
+    assert len(service.request_approvals_for_ready_signals()) == 1
+
+
+def test_breakout_now_respects_dilution_veto(store):
+    advisory = {"is_dilutive": True, "conviction": 0.9}
+    service = _service_with_catalyst(store, advisory, veto_enabled=True)
+    res = service.submit_breakout_now("DILUT", trigger=14.0, stop=13.45, last_price=14.0)
+    assert res["ok"] is False and res["skipped"] == "dilution_veto"

trading_execution.py

@@ -160,6 +160,13 @@ class ExecutionSettings:
     # than this many minutes WHILE other names are still printing (LULD halts run
     # >= 5 min, so 3 catches a real halt without tripping on normal IEX lag).
     halt_max_bar_gap_min: float = 3.0
+    # --- catalyst dilution veto (Phase 2; ships OFF) -----------------------
+    # block an entry when the LLM catalyst advisory says the move is driven by a
+    # CONFIRMED dilutive offering (a gap that fades as new shares hit). Requires
+    # a catalyst_provider AND a conviction at/above the floor below. Default OFF
+    # so the LLM stays advisory-only until its accuracy is trusted.
+    catalyst_veto_enabled: bool = False
+    catalyst_veto_min_conviction: float = 0.6
 
     @classmethod
     def from_env(cls, env: dict[str, str] | None = None) -> "ExecutionSettings":
@@ -222,6 +229,10 @@ class ExecutionSettings:
             halt_max_bar_gap_min=float(
                 values.get("TRADING_HALT_MAX_BAR_GAP_MIN", "3")
             ),
+            catalyst_veto_enabled=flag("NEWS_DILUTION_VETO_ENABLED", "0"),
+            catalyst_veto_min_conviction=float(
+                values.get("NEWS_DILUTION_VETO_CONVICTION", "0.6")
+            ),
         )
 
 
@@ -238,9 +249,14 @@ class TradingExecutionService:
         equity: float | None = None,
         price_provider=None,
         now_fn=None,
+        catalyst_provider=None,
     ):
         self.store = store
         self.settings = settings or ExecutionSettings.from_env()
+        # optional callable: symbol -> catalyst advisory dict (or None), from the
+        # LLM enrichment cache. Used ONLY for the (gated) dilution veto. Injected
+        # so the execution service never reads the DB itself.
+        self.catalyst_provider = catalyst_provider
         self.trading_mode = trading_mode or TradingModeSettings.from_env()
         self.session_id = session_id
         self.equity = equity or self.settings.default_equity
@@ -525,6 +541,46 @@ class TradingExecutionService:
         return self._halted
 
     @_locked
+    def _dilution_vetoed(self, symbol: str) -> dict | None:
+        """If the gated catalyst veto is on AND the LLM advisory flags a confirmed
+        dilutive offering at/above the conviction floor, return the advisory (so
+        the caller can emit a risk event + skip). Otherwise None. Never raises —
+        a missing/odd advisory simply doesn't veto."""
+        if not (self.settings.catalyst_veto_enabled and self.catalyst_provider):
+            return None
+        try:
+            adv = self.catalyst_provider(symbol)
+        except Exception:  # noqa: BLE001 — advisory is best-effort, never blocks on its own fault
+            return None
+        if not adv or not adv.get("is_dilutive"):
+            return None
+        try:
+            conviction = float(adv.get("conviction") or 0.0)
+        except (TypeError, ValueError):
+            return None
+        if conviction >= self.settings.catalyst_veto_min_conviction:
+            return adv
+        return None
+
+    def _emit_dilution_veto(self, symbol: str, adv: dict) -> None:
+        self.store.emit(
+            RiskRuleTriggeredEvent(
+                timestamp=datetime.now(),
+                mode=self.mode,
+                correlation_id=self.session_id,
+                message=f"{symbol}: entry blocked — confirmed dilutive catalyst",
+                rule_type="catalyst_dilution_veto",
+                rule_value=float(adv.get("conviction") or 0.0),
+                current_state={
+                    "symbol": symbol,
+                    "catalyst_type": adv.get("catalyst_type"),
+                    "sentiment": adv.get("sentiment"),
+                    "rationale": adv.get("rationale"),
+                },
+                action_taken="blocked_dilutive_offering",
+            )
+        )
+
     def request_approvals_for_ready_signals(self) -> list[str]:
         """Turn fresh ready signals into approval requests. Returns order ids."""
         if not self.settings.enabled:
@@ -562,6 +618,11 @@ class TradingExecutionService:
                     or self._in_cooldown(symbol)        # benched after recent backouts
                     or symbol in self._exited_today):   # already round-tripped today
                 continue
+            vetoed = self._dilution_vetoed(symbol)
+            if vetoed is not None:
+                self._emit_dilution_veto(symbol, vetoed)
+                self._requested_symbols.add(symbol)  # don't re-evaluate every tick
+                continue
             entry = signal.get("entry_price")
             stop = signal.get("stop_loss_price")
             if not entry or not stop or stop >= entry:
@@ -710,6 +771,10 @@ class TradingExecutionService:
             return {"ok": False, "skipped": "cooldown"}
         if symbol in self._exited_today:  # already round-tripped today — no re-entry
             return {"ok": False, "skipped": "reentry_blocked"}
+        vetoed = self._dilution_vetoed(symbol)  # gated catalyst dilution veto
+        if vetoed is not None:
+            self._emit_dilution_veto(symbol, vetoed)
+            return {"ok": False, "skipped": "dilution_veto"}
 
         # Use last_price only when it's a real positive tick. `last_price or
         # trigger` would coerce a 0.0/garbage price to the trigger — silently