momentum-system/prwire/recorder.py

"""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/<YYYY-MM-DD ET>.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()