| ' + esc(q.symbol) + ' | ' +
'' + esc(r.side || "buy") + ' | ' +
'' + riskChip(r) + ' | ' +
+ '' + catalystCell(q.catalyst) + ' | ' +
'' + esc(q.approval_mode || "manual") + ' | ' +
'' +
'' +
diff --git a/config.py b/config.py
index 908a35f..0a9c664 100644
--- a/config.py
+++ b/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):
diff --git a/research/ingestion/news_enrichment.py b/research/ingestion/news_enrichment.py
new file mode 100644
index 0000000..52c4ed7
--- /dev/null
+++ b/research/ingestion/news_enrichment.py
@@ -0,0 +1,318 @@
+"""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
+ 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
+ counts["errors"] += 1
+ logger.debug("catalyst store failed: %s", exc, exc_info=True)
+ counts["enriched"] += 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
diff --git a/run_live_paper.py b/run_live_paper.py
index 0121e6d..672c26e 100644
--- a/run_live_paper.py
+++ b/run_live_paper.py
@@ -49,6 +49,8 @@ from research.ingestion.market_data import (
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 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
@@ -194,6 +196,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(
@@ -211,6 +240,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()
@@ -254,6 +286,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
@@ -285,6 +318,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,
}
@@ -293,6 +328,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"]
@@ -559,6 +595,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.
@@ -610,7 +672,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,
})
@@ -767,7 +829,21 @@ def main(argv: list[str] | None = None) -> int:
return (f"EOD FLATTEN closed [{closed}]"
+ (f" errors={res['errors']} (will retry)" if res["errors"] else ""))
+ # Local-LLM (Ollama) news/catalyst enrichment — advisory layer, OFF the hot
+ # path. Classifies recently-ingested headlines (catalyst type / sentiment /
+ # dilution) into news_catalyst_cache for the dashboard. Gated by both the
+ # enrichment flag AND ollama.enabled; degrades to a no-op if Ollama is down.
+ def step_enrich():
+ if not (ollama_cfg.enabled and ollama_cfg.enrichment_enabled):
+ return "skipped (disabled)"
+ res = enrich_recent_news(rt["research_con"], ollama_cfg)
+ return (f"enriched={res['enriched']} skipped={res['skipped']} "
+ f"errors={res['errors']}")
+
scheduler = Scheduler()
+ scheduler.add("enrich", step_enrich,
+ float(ollama_cfg.enrichment_interval_seconds),
+ enabled=ollama_cfg.enabled and ollama_cfg.enrichment_enabled)
scheduler.add("discover", step_discover,
float(os.environ.get("DISCOVER_INTERVAL_SECONDS", "300")),
enabled=_flag("DISCOVER_ENABLED", "1"))
@@ -793,7 +869,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", "watch", "arm", "sync", "execute", "exits", "eod"):
+ for name in ("discover", "ingest", "enrich", "watch", "arm", "sync", "execute", "exits", "eod"):
for task in scheduler.tasks:
if task.name == name and task.enabled:
results.update(scheduler_run_one(task))
diff --git a/runtime/watcher.py b/runtime/watcher.py
index ab65194..56c1db5 100644
--- a/runtime/watcher.py
+++ b/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
diff --git a/storage/pg_schema.sql b/storage/pg_schema.sql
index 3e9ec4c..10b2d8e 100644
--- a/storage/pg_schema.sql
+++ b/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,
diff --git a/storage/projections.py b/storage/projections.py
index 34f7caa..085b6b1 100644
--- a/storage/projections.py
+++ b/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
diff --git a/strategy/evaluation/quality.py b/strategy/evaluation/quality.py
index cd45eea..24e74a2 100644
--- a/strategy/evaluation/quality.py
+++ b/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 = (
- 0.20 * gap_component
- + 0.25 * rvol_component
- + 0.30 * structure_component
- + 0.10 * vwap_component
- + 0.05 * opening_component
- + 0.10 * dq_component
- )
+ components = {
+ "gap": gap_component,
+ "relative_volume": rvol_component,
+ "structure": structure_component,
+ "vwap": vwap_component,
+ "opening_strength": opening_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={
- "gap": gap_component,
- "relative_volume": rvol_component,
- "structure": structure_component,
- "vwap": vwap_component,
- "opening_strength": opening_component,
- "data_quality": dq_component,
- },
+ components=components,
tradeable=score >= thresholds.min_tradeable,
)
diff --git a/strategy/evaluation/setup_evaluator.py b/strategy/evaluation/setup_evaluator.py
index 0ff9d00..ef5ba32 100644
--- a/strategy/evaluation/setup_evaluator.py
+++ b/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
diff --git a/tests/integration/test_catalyst_enrichment_flow.py b/tests/integration/test_catalyst_enrichment_flow.py
new file mode 100644
index 0000000..5949379
--- /dev/null
+++ b/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
diff --git a/tests/unit/research/test_news_enrichment.py b/tests/unit/research/test_news_enrichment.py
new file mode 100644
index 0000000..3936b76
--- /dev/null
+++ b/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"
diff --git a/tests/unit/runtime/test_watcher.py b/tests/unit/runtime/test_watcher.py
index cc45cb5..7009e63 100644
--- a/tests/unit/runtime/test_watcher.py
+++ b/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)
diff --git a/tests/unit/storage/test_catalyst_advisory_projection.py b/tests/unit/storage/test_catalyst_advisory_projection.py
new file mode 100644
index 0000000..b5af1d8
--- /dev/null
+++ b/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) == {}
diff --git a/tests/unit/strategy/test_quality_catalyst.py b/tests/unit/strategy/test_quality_catalyst.py
new file mode 100644
index 0000000..c8ed907
--- /dev/null
+++ b/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
diff --git a/tests/unit/test_setup_evaluator.py b/tests/unit/test_setup_evaluator.py
index 30acec0..bc31e93 100644
--- a/tests/unit/test_setup_evaluator.py
+++ b/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"]
diff --git a/tests/unit/test_trading_execution.py b/tests/unit/test_trading_execution.py
index 8e801b8..7b783c7 100644
--- a/tests/unit/test_trading_execution.py
+++ b/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"
diff --git a/trading_execution.py b/trading_execution.py
index e53b9af..eb94e82 100644
--- a/trading_execution.py
+++ b/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
|