momentum devandClaude Opus 5.5
98805279d1
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/<date ET>.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 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01B3PXXn2tHBzdv6rYtGPz86
60 lines
2.3 KiB
Python
60 lines
2.3 KiB
Python
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
|