quant-trading/research/lab/fetch_13d.py

"""Fetch SC 13D / 13G filings (activist and 5% passive stakes) from EDGAR.

Quarterly full-index files list every filing:

    https://www.sec.gov/Archives/edgar/full-index/YYYY/QTRn/form.idx

Each is ~40-60 MB fixed-width text. Rows for SC 13D, SC 13D/A, SC 13G and
SC 13G/A are kept (filed date, CIK, company, archive path). Amendments are
flagged via the /A suffix so an event study can use originals only.

Writes data/research/edgar_13d.csv (append-safe per quarter; a quarter
already present in the CSV is refetched only with --refresh).

    PYTHONPATH=. uv run python -m research.lab.fetch_13d [--quarters N]
"""
from __future__ import annotations

import io
import os
import re
import sys
import time

import pandas as pd
import requests

from research.lab.fetch_us import _proxy_env

OUT = os.path.join("data", "research", "edgar_13d.csv")
URL = "https://www.sec.gov/Archives/edgar/full-index/{y}/QTR{q}/form.idx"
UA = {"User-Agent": "quant-research/0.1 gahow@example.com", "Accept-Encoding": "gzip"}
ROW = re.compile(
    r"^(?P<form>.{1,16}?)\s{2,}(?P<company>.+?)\s+(?P<cik>\d{4,10})\s+"
    r"(?P<date>\d{4}-\d{2}-\d{2})\s+(?P<file>edgar/\S+)\s*$")
KEEP = ("SC 13D", "SC 13G")


def quarters() -> list[tuple[int, int]]:
    out = []
    for y in range(2016, 2027):
        for q in range(1, 5):
            if y == 2026 and q > 3:
                continue
            out.append((y, q))
    return out


def fetch_quarter(y: int, q: int) -> list[dict] | None:
    url = URL.format(y=y, q=q)
    for attempt in range(6):
        try:
            r = requests.get(url, headers=UA, timeout=300)
            r.raise_for_status()
            # SEC throttles sustained bulk downloads with small non-index pages
            if len(r.content) < 5_000_000 or "Master Index of EDGAR" not in r.text[:2000]:
                print(f"  {y}Q{q}: throttled or unexpected content "
                      f"({len(r.content)}B), backing off")
                time.sleep(20 * (attempt + 1))
                continue
            rows = []
            text = io.StringIO(r.text)
            for line in text:
                if not (line.startswith("SC 13D") or line.startswith("SC 13G")):
                    continue
                m = ROW.match(line.rstrip("\n"))
                if not m:
                    continue
                d = m.groupdict()
                if not d["form"].startswith(KEEP):
                    continue
                rows.append({"quarter": f"{y}Q{q}", "form": d["form"].strip(),
                             "company": d["company"].strip(), "cik": int(d["cik"]),
                             "filed": d["date"], "path": d["file"]})
            return rows
        except Exception as exc:  # noqa: BLE001
            if attempt == 5:
                print(f"  FAILED {y}Q{q}: {exc}")
            else:
                time.sleep(5 * (attempt + 1))
    return None


def main() -> None:
    _proxy_env()
    limit = None
    if "--quarters" in sys.argv:
        limit = int(sys.argv[sys.argv.index("--quarters") + 1])
    done: set[str] = set()
    prev = None
    if os.path.exists(OUT):
        prev = pd.read_csv(OUT)
        done = set(prev["quarter"].unique())
    if "--refresh-since" in sys.argv:
        since = sys.argv[sys.argv.index("--refresh-since") + 1].upper()
        def ge(quarter: str) -> bool:
            y, q = int(quarter[:4]), int(quarter[-1])
            sy, sq = int(since[:4]), int(since[-1])
            return (y, q) >= (sy, sq)
        if prev is not None:
            prev = prev[~prev["quarter"].apply(ge)]
        done = {q for q in done if not ge(q)}
    todo = [(y, q) for (y, q) in quarters() if f"{y}Q{q}" not in done]
    if limit:
        todo = todo[:limit]
    print(f"quarters: {len(quarters())}, stored: {len(done)}, fetching: {len(todo)}")
    frames = [prev] if prev is not None else []
    for y, q in todo:
        rows = fetch_quarter(y, q)
        if rows is None:
            continue
        df = pd.DataFrame(rows)
        if len(df) == 0:
            print(f"  {y}Q{q}: empty, skipped")
            continue
        frames.append(df)
        print(f"  {y}Q{q}: {len(df)} SC13 rows")
        pd.concat(frames).drop_duplicates(subset=["form", "cik", "filed", "path"]).to_csv(OUT, index=False)
        time.sleep(4)
    if frames:
        allrows = pd.concat(frames).drop_duplicates(subset=["form", "cik", "filed", "path"])
        allrows.to_csv(OUT, index=False)
        print(f"stored {len(allrows)} rows -> {OUT}")


if __name__ == "__main__":
    main()