From 98805279d10534f6d477e6e2135e74e116b40e5e Mon Sep 17 00:00:00 2001 From: momentum dev Date: Thu, 1 Oct 2026 08:12:10 -0500 Subject: [PATCH] feat: real-time press-release and SEC filing recorder (prwire) Premarket news timing could not be tested on history. Alpaca's Benzinga feed has almost no PR-wire items for microcaps, and wire RSS feeds only keep the latest few dozen items. So from now on, every item is recorded with our first-seen time. - Polls PR Newswire, GlobeNewswire, Business Wire and SEC EDGAR's current 8-K, 6-K and 424B feeds every 20 s, in parallel. A failing feed backs off up to 10 min. - Each new item goes to data/prwire/.jsonl with seen_utc, feed, title, link, tickers (exchange-tagged, e.g. "(Nasdaq CM: ABCD)"; SEC filings via CIK) and form. Restarts do not re-log items already recorded. - Runs on the workstation via Task Scheduler "Momentum\PrWireRecorder" (pythonw, no console window; re-fired every 10 min with IgnoreNew so it self-heals). Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01B3PXXn2tHBzdv6rYtGPz86 --- prwire/__init__.py | 1 + prwire/recorder.py | 161 +++++++++++++++++++++++++++++ prwire/run_recorder.cmd | 9 ++ tests/unit/test_prwire_recorder.py | 59 +++++++++++ 4 files changed, 230 insertions(+) create mode 100644 prwire/__init__.py create mode 100644 prwire/recorder.py create mode 100644 prwire/run_recorder.cmd create mode 100644 tests/unit/test_prwire_recorder.py diff --git a/prwire/__init__.py b/prwire/__init__.py new file mode 100644 index 0000000..a15e118 --- /dev/null +++ b/prwire/__init__.py @@ -0,0 +1 @@ +"""Real-time press-release / SEC filing recorder (premarket news-timing research).""" diff --git a/prwire/recorder.py b/prwire/recorder.py new file mode 100644 index 0000000..35d4c12 --- /dev/null +++ b/prwire/recorder.py @@ -0,0 +1,161 @@ +"""Record every press release and SEC 8-K / 6-K / 424B filing the moment we first see it. + +Why: premarket news timing could not be tested on history. The Benzinga feed carries almost no PR-wire items for +microcaps, and wire RSS feeds only show the latest few dozen items. From now on each item is logged with OUR +first-seen time (UTC) and the tickers it names, so later we can ask: "a fresh release hits the wire and the move +starts N minutes later; is that tradable?" + +usage: python -m prwire.recorder (runs until killed; Task Scheduler re-fires it if it dies) +out: data/prwire/.jsonl, one JSON object per new item: + seen_utc, feed, title, link, guid, published, tickers, cik, form, snippet +""" +from __future__ import annotations + +import hashlib +import json +import os +import re +import sys +import time +import urllib.request +from concurrent.futures import ThreadPoolExecutor +from datetime import datetime, timezone +from pathlib import Path +from zoneinfo import ZoneInfo + +ROOT = Path(__file__).resolve().parent.parent +sys.path.insert(0, str(ROOT)) +from research.ingestion.rss import parse_rss # noqa: E402 +from research.ingestion.sec_dilution import USER_AGENT # noqa: E402 + +ET = ZoneInfo("America/New_York") +OUT = ROOT / "data" / "prwire" +POLL_S = float(os.environ.get("PRWIRE_POLL_S", "20")) +_SEC = ("https://www.sec.gov/cgi-bin/browse-edgar?action=getcurrent&type={t}&company=&dateb=&owner=include" + "&start=0&count=100&output=atom") +FEEDS = { + "prnewswire": "https://www.prnewswire.com/rss/news-releases-list.rss", + "globenewswire": "https://www.globenewswire.com/RssFeed/feedTitle/GlobeNewswire%20-%20All%20News", + "businesswire": "https://feed.businesswire.com/rss/home/?rss=G1QFDERJXkJeEFtRWA==", + "sec_8k": _SEC.format(t="8-K"), + "sec_6k": _SEC.format(t="6-K"), + "sec_424b": _SEC.format(t="424B"), +} +# "(Nasdaq: ABCD)", "NASDAQ CM: ABCD", "NYSE American: XYZ", "OTCQB: ABCD", "Nasdaq:ABCD" ... +TICKER_RE = re.compile( + r"\b(?:NASDAQ|Nasdaq|NYSE(?:\s+(?:American|MKT|Arca))?|AMEX|OTCQ[BX]|OTC(?:\s+Markets)?|OTC\s+Pink|Cboe)" + r"(?:\s*(?:CM|GM|GS|Capital\s+Market|Global\s+(?:Select\s+)?Market))?\s*[:\-]\s*\$?([A-Z]{1,5}(?:\.[A-Z]{1,2})?)\b") +CIK_RE = re.compile(r"\((\d{10})\)") +FORM_RE = re.compile(r"^\s*([0-9A-Z\-/]+)\s+-\s+") + + +def extract_tickers(text: str) -> list[str]: + return sorted(set(TICKER_RE.findall(text or ""))) + + +def _cik_map() -> dict[int, str]: + f = ROOT / "data" / "ross" / "judge" / "edgar_cache" / "company_tickers.json" + if not f.exists() or time.time() - f.stat().st_mtime > 7 * 86400: + try: + req = urllib.request.Request("https://www.sec.gov/files/company_tickers.json", + headers={"User-Agent": USER_AGENT}) + with urllib.request.urlopen(req, timeout=30) as r: + f.parent.mkdir(parents=True, exist_ok=True) + f.write_bytes(r.read()) + except Exception: # noqa: BLE001 + if not f.exists(): + return {} + return {int(v["cik_str"]): v["ticker"].upper() for v in json.loads(f.read_text()).values()} + + +def item_key(feed: str, it: dict) -> str: + return hashlib.sha1(f"{feed}|{it.get('url') or ''}|{it.get('title') or ''}".encode()).hexdigest() + + +def to_record(feed: str, it: dict, seen_utc: str, ciks: dict[int, str]) -> dict: + title, snippet = it.get("title") or "", it.get("snippet") or "" + rec = dict(seen_utc=seen_utc, feed=feed, title=title, link=it.get("url"), guid=item_key(feed, it), + published=it.get("published"), tickers=extract_tickers(f"{title} {snippet}"), cik=None, form=None, + snippet=snippet[:400]) + if feed.startswith("sec_"): + m = CIK_RE.search(title) + if m: + rec["cik"] = int(m.group(1)) + t = ciks.get(rec["cik"]) + rec["tickers"] = [t] if t else [] + f = FORM_RE.match(title) + rec["form"] = f.group(1) if f else None + return rec + + +def fetch(url: str) -> list[dict]: + ua = USER_AGENT if "sec.gov" in url else "Mozilla/5.0 (Windows NT 10.0; Win64; x64) momentum-research/1.0" + req = urllib.request.Request(url, headers={"User-Agent": ua, "Accept": "application/rss+xml,application/xml,*/*"}) + with urllib.request.urlopen(req, timeout=15) as r: + return parse_rss(r.read()) + + +class Recorder: + def __init__(self): + OUT.mkdir(parents=True, exist_ok=True) + self.ciks = _cik_map() + self.seen: set[str] = set() + for f in sorted(OUT.glob("*.jsonl"))[-3:]: # don't re-log items already recorded before a restart + for line in f.read_text(encoding="utf-8").splitlines(): + try: + g = json.loads(line).get("guid") + except ValueError: + continue + if g: + self.seen.add(g) + self.errors: dict[str, int] = {} + self.skip_until: dict[str, float] = {} + self.pool = ThreadPoolExecutor(max(1, len(FEEDS))) + + def poll_once(self, fetcher=fetch, now: datetime | None = None) -> int: + """Fetch every due feed in parallel (one slow feed never delays the rest) and log the new items. + A failing feed backs off 20 s, 40 s, 80 s... up to 10 min instead of being hit every poll.""" + now = now or datetime.now(timezone.utc) + seen_utc = now.isoformat(timespec="seconds") + day = now.astimezone(ET).strftime("%Y-%m-%d") + mono = time.monotonic() + due = {f: u for f, u in FEEDS.items() if self.skip_until.get(f, 0.0) <= mono} + + def get(url): + try: + return fetcher(url), None + except Exception as e: # noqa: BLE001 + return None, e + + results = dict(zip(due, self.pool.map(get, due.values()))) + new = 0 + with open(OUT / f"{day}.jsonl", "a", encoding="utf-8") as fh: + for feed, (items, err) in results.items(): + if err is not None: + n = self.errors[feed] = self.errors.get(feed, 0) + 1 + self.skip_until[feed] = mono + min(600.0, POLL_S * 2 ** (n - 1)) + if n in (1, 10, 100): # log an outage, not every failed poll + fh.write(json.dumps(dict(seen_utc=seen_utc, feed=feed, error=str(err)[:200], guid=None)) + "\n") + continue + if self.errors.get(feed): + fh.write(json.dumps(dict(seen_utc=seen_utc, feed=feed, recovered_after=self.errors[feed], + guid=None)) + "\n") + self.errors[feed] = 0 + for it in items: + k = item_key(feed, it) + if k in self.seen: + continue + self.seen.add(k) + fh.write(json.dumps(to_record(feed, it, seen_utc, self.ciks)) + "\n") + new += 1 + return new + + def run(self): + while True: + t0 = time.time() + self.poll_once() + time.sleep(max(1.0, POLL_S - (time.time() - t0))) + + +if __name__ == "__main__": + Recorder().run() diff --git a/prwire/run_recorder.cmd b/prwire/run_recorder.cmd new file mode 100644 index 0000000..e217d57 --- /dev/null +++ b/prwire/run_recorder.cmd @@ -0,0 +1,9 @@ +@echo off +rem Press-release / SEC filing recorder. Task Scheduler "Momentum\PrWireRecorder" re-fires this every 10 min +rem (IgnoreNew), so it restarts by itself if it dies. Output: data\prwire\YYYY-MM-DD.jsonl +setlocal +cd /d "%~dp0.." +set PYTHONIOENCODING=utf-8 +if not exist data\prwire mkdir data\prwire +data\ross\.venv\Scripts\python.exe -m prwire.recorder >> "data\prwire\recorder.log" 2>&1 +exit /b %ERRORLEVEL% diff --git a/tests/unit/test_prwire_recorder.py b/tests/unit/test_prwire_recorder.py new file mode 100644 index 0000000..f02d949 --- /dev/null +++ b/tests/unit/test_prwire_recorder.py @@ -0,0 +1,59 @@ +from datetime import datetime, timezone + +import pytest + +from prwire import recorder +from prwire.recorder import Recorder, extract_tickers, to_record + + +@pytest.mark.parametrize("text,want", [ + ("Acme Corp (NASDAQ: ACME) announces FDA clearance", ["ACME"]), + ("Beta Inc. (Nasdaq CM: BETA) and Gamma (NYSE American: GMA)", ["BETA", "GMA"]), + ("Delta (OTCQB: DLTA) signs deal", ["DLTA"]), + ("Epsilon Ltd. (NASDAQ:EPSN)", ["EPSN"]), + ("No exchange mention here, just $SPY chatter", []), +]) +def test_extract_tickers(text, want): + assert extract_tickers(text) == want + + +def test_sec_item_maps_cik_to_ticker_and_form(): + it = {"title": "8-K - Dare Bioscience, Inc. (0001401914) (Filer)", "url": "u", "published": "p", "snippet": ""} + rec = to_record("sec_8k", it, "2026-10-01T13:00:00+00:00", {1401914: "DARE"}) + assert rec["tickers"] == ["DARE"] and rec["form"] == "8-K" and rec["cik"] == 1401914 + + +def test_poll_logs_each_item_once_and_survives_a_feed_error(tmp_path, monkeypatch): + monkeypatch.setattr(recorder, "OUT", tmp_path) + monkeypatch.setattr(recorder, "_cik_map", lambda: {}) + monkeypatch.setattr(recorder, "FEEDS", {"a": "A", "b": "B"}) + items = [{"title": "X (NASDAQ: XX) news", "url": "l1", "published": "p", "snippet": ""}] + + def fetcher(url): + if url == "B": + raise OSError("down") + return items + + r = Recorder() + now = datetime(2026, 10, 1, 12, 0, tzinfo=timezone.utc) + assert r.poll_once(fetcher, now) == 1 + assert r.poll_once(fetcher, now) == 0 # same item again: not re-logged + lines = (tmp_path / "2026-10-01.jsonl").read_text().splitlines() + assert len(lines) == 2 # the item + one error row for feed b + assert Recorder().seen == r.seen # a restart reloads what was already recorded + + +def test_failing_feed_backs_off(tmp_path, monkeypatch): + monkeypatch.setattr(recorder, "OUT", tmp_path) + monkeypatch.setattr(recorder, "_cik_map", lambda: {}) + monkeypatch.setattr(recorder, "FEEDS", {"bad": "B"}) + calls = [] + + def fetcher(url): + calls.append(url) + raise OSError("timeout") + + r = Recorder() + r.poll_once(fetcher) + r.poll_once(fetcher) # still inside the back-off window: not fetched again + assert calls == ["B"] and r.errors["bad"] == 1