#!/usr/bin/env python3 """Collect historical headlines from Google News RSS into data/history.db. One request per (query, date window), e.g. ``gold site:reuters.com after:2019-07-22 before:2019-07-29``. Google returns at most 100 items per request. The crawl is resumable: finished (query, window) pairs are recorded in ``crawl_log`` and skipped on the next run. Requests are sequential with a fixed delay; on HTTP 429/503 the crawler backs off and stops after repeated refusals rather than hammering the service. Headlines are stored once per normalised text (``hash``); the queries that found a headline are kept in ``queries``. Nothing here labels or filters against the golden set; label_history.py and the trainer do. Usage: python data_collect/crawl_google_news.py # precise set python data_collect/crawl_google_news.py --query-set broad # off-topic side python data_collect/crawl_google_news.py --query-set growth --window-days 30 # growth data + equities python data_collect/crawl_google_news.py --limit 20 # smoke test """ from __future__ import annotations import argparse import email.utils import hashlib import html import random import re import sqlite3 import sys import time import urllib.error import urllib.parse import urllib.request import xml.etree.ElementTree as ET from datetime import date, datetime, timedelta, timezone from pathlib import Path ML_NEWS_DIR = Path(__file__).resolve().parent.parent DEFAULT_DB = ML_NEWS_DIR / "data" / "history.db" FEED_URL = "https://news.google.com/rss/search?q={q}&hl=en-US&gl=US&ceid=US:en" USER_AGENT = "Mozilla/5.0 (X11; Linux x86_64) ML-NEWS research crawler" #: Scoped to financial publishers: unscoped queries return mostly #: general news ("gold price" -> phone prices, "safe haven" -> baby-box #: laws). Reuters answers each weekly window with up to 100 market #: headlines. Topics cover gold doctrine (Fed, dollar, yields, #: inflation, jobs, central banks, geopolitics, risk appetite) plus #: off-topic controls (oil, crypto, single stocks) so the relevance head #: sees news that does not move gold. BROAD_QUERIES = ( "gold site:reuters.com", "Fed site:reuters.com", "dollar site:reuters.com", "yields site:reuters.com", "inflation site:reuters.com", "payrolls OR jobless site:reuters.com", "\"central bank\" site:reuters.com", "stocks site:reuters.com", "war OR sanctions OR tensions site:reuters.com", "euro OR yen OR sterling site:reuters.com", "oil site:reuters.com", "bitcoin OR crypto site:reuters.com", "gold site:cnbc.com", "markets site:cnbc.com", "site:marketwatch.com", ) #: A keyword in the query matches anywhere in the article, so the broad #: set is ~93% headlines that do not move gold (2026-09-23 pilot: 70 of #: 1,000 DeepSeek-relevant). ``intitle:`` restricts the match to the #: headline: 3-15 items per week per query, mostly gold doctrine. The #: broad set supplies the not-relevant side, this one the relevant side. PRECISE_QUERIES = tuple( f"intitle:{term} site:{site}" for site in ("reuters.com", "cnbc.com", "marketwatch.com") for term in ("gold", "bullion", "Fed", "Powell", "dollar", "yields", "inflation", "payrolls", '"rate cut"', '"rate hike"', '"safe haven"') if site == "reuters.com" or term in ("gold", "Fed", "dollar", "yields", "inflation") ) #: The two golden-set categories the 2026-09-27 model still fails are #: strong-US-data (3/7) and risk-on (1/4): "GDP beats", "stocks at record #: highs", "ceasefire lifts markets" should all be bearish for gold, but #: no precise query covers growth data or equities. ``intitle:`` again, #: so label_history.py puts these in the precise pool. GROWTH_TERMS = ("GDP", '"retail sales"', "ISM", "PMI", '"consumer confidence"', '"jobless claims"', '"consumer spending"', "manufacturing") RISK_TERMS = ('"record high"', '"risk appetite"', '"Wall Street"', '"S&P 500"', "ceasefire", '"trade deal"', "equities") GROWTH_QUERIES = tuple( f"intitle:{term} site:{site}" for site in ("reuters.com", "cnbc.com") for term in GROWTH_TERMS + RISK_TERMS if site == "reuters.com" or term in ("GDP", '"retail sales"', '"jobless claims"', '"record high"', '"S&P 500"') ) QUERY_SETS = {"broad": BROAD_QUERIES, "precise": PRECISE_QUERIES, "growth": GROWTH_QUERIES} SCHEMA = """ CREATE TABLE IF NOT EXISTS headlines ( hash TEXT PRIMARY KEY, headline TEXT NOT NULL, raw_title TEXT NOT NULL, publisher TEXT, published_at TEXT, first_query TEXT NOT NULL, queries TEXT NOT NULL, link TEXT, crawled_at TEXT NOT NULL ); CREATE INDEX IF NOT EXISTS idx_headlines_published ON headlines (published_at); CREATE TABLE IF NOT EXISTS crawl_log ( query TEXT NOT NULL, window_start TEXT NOT NULL, items INTEGER NOT NULL, new_headlines INTEGER NOT NULL, crawled_at TEXT NOT NULL, PRIMARY KEY (query, window_start) ); """ _SPACE = re.compile(r"\s+") def normalise(text: str) -> str: return _SPACE.sub(" ", html.unescape(text)).strip() def headline_key(headline: str) -> str: """Hash of the case-folded, whitespace-collapsed headline.""" return hashlib.sha256(normalise(headline).casefold().encode("utf-8")).hexdigest() def strip_publisher(title: str, publisher: str | None) -> str: """Google appends ' - Publisher' to every title; drop it.""" title = normalise(title) if publisher: suffix = f" - {normalise(publisher)}" if title.endswith(suffix): return title[: -len(suffix)].strip() return title def windows_of(start: date, end: date, days: int): day = start while day < end: yield day, min(day + timedelta(days=days), end) day += timedelta(days=days) def fetch(url: str, timeout: float) -> bytes: request = urllib.request.Request(url, headers={"User-Agent": USER_AGENT}) with urllib.request.urlopen(request, timeout=timeout) as response: return response.read() def parse_items(payload: bytes): root = ET.fromstring(payload) for item in root.findall("./channel/item"): title = item.findtext("title") or "" publisher = item.findtext("source") published = item.findtext("pubDate") published_at = None if published: try: published_at = email.utils.parsedate_to_datetime(published).astimezone( timezone.utc).isoformat() except (TypeError, ValueError): published_at = None headline = strip_publisher(title, publisher) if headline: yield { "headline": headline, "raw_title": normalise(title), "publisher": publisher, "published_at": published_at, "link": item.findtext("link"), } def store(conn: sqlite3.Connection, query: str, items) -> tuple[int, int]: now = datetime.now(timezone.utc).isoformat() seen = new = 0 for it in items: seen += 1 key = headline_key(it["headline"]) row = conn.execute("SELECT queries FROM headlines WHERE hash = ?", (key,)).fetchone() if row is None: conn.execute( "INSERT INTO headlines (hash, headline, raw_title, publisher, published_at," " first_query, queries, link, crawled_at) VALUES (?,?,?,?,?,?,?,?,?)", (key, it["headline"], it["raw_title"], it["publisher"], it["published_at"], query, query, it["link"], now)) new += 1 elif query not in row[0].split("|"): conn.execute("UPDATE headlines SET queries = ? WHERE hash = ?", (row[0] + "|" + query, key)) return seen, new def main() -> int: ap = argparse.ArgumentParser(description=__doc__.split("\n\n")[0]) ap.add_argument("--db", type=Path, default=DEFAULT_DB) ap.add_argument("--start", type=date.fromisoformat, default=date(2019, 1, 7)) ap.add_argument("--end", type=date.fromisoformat, default=date(2026, 9, 1)) ap.add_argument("--delay", type=float, default=2.0, help="seconds between requests") ap.add_argument("--timeout", type=float, default=20.0) ap.add_argument("--limit", type=int, default=0, help="stop after N requests (0 = all)") ap.add_argument("--query-set", choices=sorted(QUERY_SETS), default="precise") ap.add_argument("--window-days", type=int, default=0, help="search window length (default: 7 broad, 14 precise; precise " "queries return 3-15 items a week, far under the 100 cap)") ap.add_argument("--max-refusals", type=int, default=5, help="stop after this many consecutive 429/503 responses") args = ap.parse_args() args.db.parent.mkdir(parents=True, exist_ok=True) conn = sqlite3.connect(args.db) conn.executescript(SCHEMA) done = {(q, w) for q, w in conn.execute("SELECT query, window_start FROM crawl_log")} queries = QUERY_SETS[args.query_set] days = args.window_days or (14 if args.query_set == "precise" else 7) windows = list(windows_of(args.start, args.end, days)) # Seeded shuffle, window-major: an interrupted crawl still covers # every year and every query evenly, so a pilot on a partial crawl # is not all 2019. random.Random(0).shuffle(windows) todo = [(q, a, b) for a, b in windows for q in queries if (q, a.isoformat()) not in done] print(f"{args.query_set}: {len(windows)} windows of {days} days x {len(queries)} queries; " f"{len(done)} done, {len(todo)} to go", flush=True) requests = refusals = 0 started = time.monotonic() for query, a, b in todo: if args.limit and requests >= args.limit: break q = f"{query} after:{a.isoformat()} before:{b.isoformat()}" url = FEED_URL.format(q=urllib.parse.quote_plus(q)) try: payload = fetch(url, args.timeout) except urllib.error.HTTPError as error: if error.code in (429, 503): refusals += 1 wait = min(600, 30 * 2 ** (refusals - 1)) print(f"HTTP {error.code} on {q!r}; refusal {refusals}, sleeping {wait}s", flush=True) if refusals >= args.max_refusals: print("too many refusals, stopping; rerun later to resume", flush=True) break time.sleep(wait) continue print(f"HTTP {error.code} on {q!r}; skipped", flush=True) time.sleep(args.delay) continue except (urllib.error.URLError, TimeoutError, OSError) as error: print(f"network error on {q!r}: {error}; skipped", flush=True) time.sleep(args.delay * 5) continue refusals = 0 requests += 1 try: seen, new = store(conn, query, parse_items(payload)) except ET.ParseError as error: print(f"bad XML for {q!r}: {error}; skipped", flush=True) time.sleep(args.delay) continue conn.execute( "INSERT OR REPLACE INTO crawl_log VALUES (?,?,?,?,?)", (query, a.isoformat(), seen, new, datetime.now(timezone.utc).isoformat())) conn.commit() if requests % 100 == 0: total = conn.execute("SELECT count(*) FROM headlines").fetchone()[0] rate = requests / (time.monotonic() - started) print(f"{requests} requests, window {a}, {total} unique headlines, " f"{rate * 60:.0f} req/min", flush=True) time.sleep(args.delay) total = conn.execute("SELECT count(*) FROM headlines").fetchone()[0] print(f"done: {requests} requests this run, {total} unique headlines in {args.db}", flush=True) conn.close() return 0 if __name__ == "__main__": sys.exit(main())