News/catalyst ranking + nightly cron for the EOD replay

News (you were right it's relevant): recent_news_map() pulls fresh catalysts per
ticker from the ingested RSS feeds; step_arm tags each armed candidate and BOOSTS
catalyst-backed gappers up the rank (NEWS_RANK_BOOST) — a fresh PR/FDA/offering
is why a small-cap runs, so it surfaces the hotter name earlier and higher. The
board shows a 📰 marker + the headline. Feeds verified live (PR Newswire /
Business Wire / GlobeNewswire) — pulling real catalysts (FDA approvals, offerings).
Forward-tested only (news isn't OOS-backtestable).

eod_replay.sh + crontab: replay at 16:05 (grow dataset), tune at 17:30.

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

eod_replay.sh

@@ -0,0 +1,11 @@
+#!/usr/bin/env bash
+# End-of-day whole-market replay: what today's logic would have done + grow the
+# stored dataset for the tuner/verification. Schedule right after the close,
+# before nightly_tune (which then tunes on the freshly-grown data).
+#   crontab -e  ->  5 16 * * 1-5  /mnt/c/code/momentum/eod_replay.sh
+ROOT="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
+VENV_PY="/home/philip/.venvs/momentum/bin/python"
+cd "$ROOT" || exit 1
+mkdir -p "$ROOT/data"
+echo "=== eod_replay $(date) ===" >> "$ROOT/data/eod_replay.log"
+PYTHONPATH="$ROOT" "$VENV_PY" eod_replay.py >> "$ROOT/data/eod_replay.log" 2>&1

momentum_cli.py

@@ -471,12 +471,15 @@ def _render_board(con):
                  ("gap", "right"), ("rvol", "right"), ("trigger", "right"),
                  ("→trigger", "right"), ("stop", "right")):
         tt.add_column(c, justify=j, no_wrap=True)
+    tt.add_column("catalyst", justify="left", overflow="ellipsis", max_width=34)
     for r in trig_rows:
         style, label = _STATE_STYLE.get(r.get("state"), ("white", r.get("state", "?")))
         price = r.get("price"); trig = r.get("trigger"); dist = r.get("dist")
         stop = r.get("stop"); gap = r.get("gap"); rvol = r.get("rvol")
+        cat = r.get("catalyst") or ""
+        sym = r.get("symbol")
         tt.add_row(
-            f"[{style}]{r.get('symbol')}[/{style}]",
+            f"[{style}]{'📰' if cat else ''}{sym}[/{style}]",
             f"[{style}]{label}[/{style}]",
             f"{price:.2f}" if isinstance(price, (int, float)) else "—",
             f"{gap:+.1f}%" if isinstance(gap, (int, float)) else "—",
@@ -485,10 +488,11 @@ def _render_board(con):
             (f"[green]▲{dist * 100:+.2f}%[/green]" if isinstance(dist, (int, float)) and dist >= 0
              else (f"{dist * 100:+.2f}%" if isinstance(dist, (int, float)) else "—")),
             f"{stop:.2f}" if isinstance(stop, (int, float)) else "—",
+            f"[cyan]{cat}[/cyan]" if cat else "[dim]—[/dim]",
         )
     if not trig_rows:
         tt.add_row("[dim]—[/dim]", "[dim]waiting for the open / first bars[/dim]",
-                   "—", "—", "—", "—", "—", "—")
+                   "—", "—", "—", "—", "—", "—", "—")
 
     sig = con.execute("SELECT timestamp, message FROM events WHERE event_type='signal_ready' "
                       "ORDER BY timestamp DESC LIMIT 5").fetchall()

research/ingestion/discovery.py

@@ -145,6 +145,32 @@ def news_candidates(
     return out[:limit]
 
 
+def recent_news_map(con, lookback_hours: int = 8) -> dict[str, str]:
+    """{ticker: latest_headline} for tickers mentioned in news within the window.
+
+    Used to PRIORITISE armed candidates that have a fresh catalyst — the reason a
+    small-cap gaps and runs. Read-only (the discovery pass does the ingest);
+    no-op ({}) if no news feeds / no rows.
+    """
+    cutoff = datetime.utcnow() - timedelta(hours=lookback_hours)
+    try:
+        rows = con.execute(
+            "SELECT raw_tickers, raw_title FROM raw_news_items "
+            "WHERE raw_tickers <> '' AND fetched_at >= ? ORDER BY fetched_at DESC LIMIT 800",
+            [cutoff],
+        ).fetchall()
+    except Exception as exc:  # noqa: BLE001
+        logger.warning("recent_news_map query failed: %s", exc)
+        return {}
+    out: dict[str, str] = {}
+    for raw_tickers, title in rows:
+        for ticker in (raw_tickers or "").split(","):
+            t = ticker.strip().upper()
+            if t and t not in out:  # rows are newest-first -> keep the latest headline
+                out[t] = (title or "").strip()[:90]
+    return out
+
+
 def run_discovery(
     store,
     research_con,

run_live_paper.py

@@ -460,10 +460,15 @@ def main(argv: list[str] | None = None) -> int:
         """Rank the most-promising gappers and pre-compute each one's opening-
         range breakout trigger/stop, so step_trigger can fire on a live cross."""
         from research import query as rq
+        from research.ingestion.discovery import recent_news_map
         from research.ingestion.signals import scan_gappers
         from strategy.evaluation.structure import opening_range
 
         session_date = _now_session_date()
+        # fresh catalysts (the discover step ingests the feeds; we just read them)
+        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"))
         candidates: list[dict] = []
         # rank ALL session symbols by gap*rvol (thresholds applied later, per
         # trigger, so the board still shows the field pre-gap)
@@ -477,6 +482,7 @@ def main(argv: list[str] | None = None) -> int:
             bars = rq.query_minute_bars(rt["research_con"], g.symbol, session_date)
             hi, lo, complete = opening_range(bars, orb_bars=orb_bars_v)
             stop = lo * (1.0 - stop_cushion_v) if lo else None
+            catalyst = news_map.get(g.symbol, "")
             candidates.append({
                 "symbol": g.symbol,
                 "gap_pct": g.gap_pct,
@@ -485,6 +491,9 @@ def main(argv: list[str] | None = None) -> int:
                 "stop": stop,
                 "range_pct": ((hi - lo) / hi) if (hi and lo and hi > 0) else 0.0,
                 "cum_volume": g.cumulative_volume,
+                "catalyst": catalyst,
+                # a fresh catalyst is WHY a small-cap runs — boost it up the rank
+                "_score": g.gap_pct * max(g.relative_volume, 0.1) * (news_boost if catalyst else 1.0),
                 "complete": complete,
             })
         # pre-open / before any gappers form: surface the queued watchlist so the
@@ -494,9 +503,12 @@ def main(argv: list[str] | None = None) -> int:
                 candidates.append({"symbol": s, "gap_pct": 0.0, "rvol": 0.0,
                                    "trigger": None, "stop": None,
                                    "range_pct": 0.0, "complete": False})
+        # catalyst-backed gappers float to the top of the armed set
+        candidates.sort(key=lambda c: c.get("_score", 0.0), reverse=True)
         book.arm(candidates)
         armed = sum(1 for t in book.triggers.values() if t.state == "armed")
-        return f"tracking={len(book.triggers)} armed={armed}"
+        n_news = sum(1 for c in candidates if c.get("catalyst"))
+        return f"tracking={len(book.triggers)} armed={armed}" + (f" news={n_news}" if n_news else "")
 
     # The trigger runs on its OWN thread, not the scheduler, so a slow ingest
     # can never freeze breakout detection. It only touches the (thread-safe)

runtime/triggers.py

@@ -59,6 +59,7 @@ class ArmedTrigger:
     range_pct: float = 0.0        # (high-low)/high — how wide the opening range is
     price: float | None = None    # latest live trade price
     cum_volume: float = 0.0       # session cumulative volume (for liquidity sizing)
+    catalyst: str = ""            # fresh news headline (if any) — why it's hot
     state: str = WAITING
     rank: int = 0
 
@@ -85,6 +86,7 @@ class ArmedTrigger:
             "price": round(self.price, 4) if self.price else None,
             "dist": round(d, 5) if d is not None else None,
             "range_pct": round(self.range_pct, 4),
+            "catalyst": self.catalyst,
         }
 
 
@@ -139,6 +141,7 @@ class ArmedTriggerBook:
             t.stop = c.get("stop")
             t.range_pct = float(c.get("range_pct") or 0.0)
             t.cum_volume = float(c.get("cum_volume") or 0.0)
+            t.catalyst = c.get("catalyst") or ""
             t.rank = rank
             if not c.get("complete") or t.trigger is None:
                 t.state = WAITING