Add local-LLM (Ollama) news/catalyst scoring

Leverage a local Ollama model to interpret news catalysts for the small-cap
momentum strategy — distinguishing bullish catalysts (FDA approval, M&A) from
dilutive offerings the technicals alone can't see. Advisory-first, then automate.

Phase 1 (advisory; no change to trade behavior):
- research/ingestion/news_enrichment.py: stdlib urllib call to Ollama with
  structured JSON output, classifying each headline into a catalyst type,
  sentiment, conviction and dilution flag. Graceful degradation (None on any
  failure) so trading is never blocked.
- news_catalyst_cache table + populate the previously-dead news_events.sentiment.
- Runs on the existing interval Scheduler (off the hot path), gated by
  OllamaConfig; deduped by the raw_news_items payload hash.
- Surface the advisory on the dashboard approval queue (projection + api + UI).

Phase 2 (automation; ships gated OFF):
- Optional catalyst_score blended into calculate_setup_quality (None reproduces
  the exact legacy weights); pure pass-through via evaluate_setup.
- Dilution veto in TradingExecutionService (slow + fast paths) emitting a
  catalyst_dilution_veto risk event, behind NEWS_DILUTION_VETO_ENABLED.

Config (OllamaConfig.from_env), .env.example docs, and tests for parsing,
persistence/dedupe, the projection, the score blend, the veto, and an
end-to-end enrich->advisory flow (Ollama mocked). Default model moves to
qwen2.5:7b-instruct (best small-model JSON, fits a 12GB GPU).

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

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