From 906237bfe305ebe2ea44a08f15b7414d1e430e35 Mon Sep 17 00:00:00 2001 From: momentum dev Date: Wed, 17 Jun 2026 17:14:57 -0400 Subject: [PATCH] News/catalyst ranking + nightly cron for the EOD replay MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 --- eod_replay.sh | 11 +++++++++++ momentum_cli.py | 8 ++++++-- research/ingestion/discovery.py | 26 ++++++++++++++++++++++++++ run_live_paper.py | 14 +++++++++++++- runtime/triggers.py | 3 +++ 5 files changed, 59 insertions(+), 3 deletions(-) create mode 100644 eod_replay.sh diff --git a/eod_replay.sh b/eod_replay.sh new file mode 100644 index 0000000..4bb8901 --- /dev/null +++ b/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 diff --git a/momentum_cli.py b/momentum_cli.py index 5b55bb2..492abaa 100644 --- a/momentum_cli.py +++ b/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() diff --git a/research/ingestion/discovery.py b/research/ingestion/discovery.py index db24196..97c9e99 100644 --- a/research/ingestion/discovery.py +++ b/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, diff --git a/run_live_paper.py b/run_live_paper.py index 18418ed..28f3843 100644 --- a/run_live_paper.py +++ b/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) diff --git a/runtime/triggers.py b/runtime/triggers.py index d394955..e5d8ff8 100644 --- a/runtime/triggers.py +++ b/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