Add local-LLM (Ollama) news/catalyst scoring #2
No files matched your search
@@ -39,6 +39,29 @@ Alpha_Vantage_Api_Key=
|
||||
# added to the watchlist. Unset = disabled (no effect on the loop).
|
||||
# NEWS_FEEDS=prnewswire=https://www.prnewswire.com/rss/news-releases-list.rss,businesswire=https://feed.businesswire.com/rss/home/?rss=G1QFDERJXkJeGVtRWQ==
|
||||
|
||||
# ---- 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)
|
||||
# ============================================================
|
||||
|
||||
+11
-1
@@ -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
|
||||
|
||||
@@ -470,6 +470,19 @@ function riskChip(req) {
|
||||
return '<span class="risk">risk <b>' + money(dollarRisk) + '</b> · ' + qty + 'sh · notional <b>' + money(notional) + '</b> · 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>";
|
||||
@@ -664,12 +677,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>' +
|
||||
|
||||
@@ -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):
|
||||
|
||||
@@ -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
|
||||
+78
-2
@@ -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))
|
||||
|
||||
+15
-1
@@ -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
|
||||
|
||||
|
||||
@@ -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,
|
||||
|
||||
@@ -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
|
||||
@@ -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,
|
||||
)
|
||||
@@ -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
|
||||
|
||||
@@ -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
|
||||
@@ -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"
|
||||
@@ -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)
|
||||
|
||||
@@ -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) == {}
|
||||
@@ -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
|
||||
@@ -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"]
|
||||
@@ -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"
|
||||
@@ -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
|
||||
|
||||
Reference in new issue
Block a user