diff --git a/.env.example b/.env.example index c6351a0..d04e409 100644 --- a/.env.example +++ b/.env.example @@ -51,7 +51,14 @@ Alpha_Vantage_Api_Key= # 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 — ships OFF): block confirmed dilutive offerings before sizing. +# 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 diff --git a/config.py b/config.py index 41d86ca..0a9c664 100644 --- a/config.py +++ b/config.py @@ -166,6 +166,9 @@ class OllamaConfig(BaseModel): # 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": @@ -205,6 +208,7 @@ class OllamaConfig(BaseModel): 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), ) diff --git a/run_live_paper.py b/run_live_paper.py index 6b9f30c..672c26e 100644 --- a/run_live_paper.py +++ b/run_live_paper.py @@ -196,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( @@ -213,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() @@ -248,24 +278,6 @@ def build_runtime(args: argparse.Namespace) -> dict: _price_cache[symbol] = (price, now) return price - # Catalyst advisory provider for the (gated) dilution veto. TTL-cached so a - # guard pass doesn't re-query per symbol; returns None on any fault so the - # veto simply doesn't fire rather than blocking entries on a DB blip. - _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()) - execution = ( TradingExecutionService( store, @@ -306,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, } @@ -314,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"] @@ -580,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. @@ -631,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, }) @@ -792,8 +833,6 @@ def main(argv: list[str] | None = None) -> int: # 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. - ollama_cfg = OllamaConfig.from_env() - def step_enrich(): if not (ollama_cfg.enabled and ollama_cfg.enrichment_enabled): return "skipped (disabled)" @@ -830,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/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)