scripts/build_sec_company_submissions_sample.py

import csv
import json
import gzip
import time
import urllib.request
from pathlib import Path

OUT_DIR = Path("data/sec")
OUT_DIR.mkdir(parents=True, exist_ok=True)

# Small high-value public company sample.
# CIK must be zero-padded to 10 digits for SEC submissions API.
COMPANIES = {
    "AAPL": "0000320193",
    "MSFT": "0000789019",
    "GOOGL": "0001652044",
    "AMZN": "0001018724",
    "NVDA": "0001045810",
    "META": "0001326801",
    "TSLA": "0001318605",
    "JPM": "0000019617",
    "WMT": "0000104169",
    "V": "0001403161",
}

USER_AGENT = "Realigns Inc support@realignsinc.com"

def fetch_json(url):
    request = urllib.request.Request(
        url,
        headers={
            "User-Agent": USER_AGENT,
            "Accept-Encoding": "gzip",
            "Host": "data.sec.gov",
        },
    )
    with urllib.request.urlopen(request, timeout=60) as response:
        raw = response.read()
        encoding = response.headers.get("Content-Encoding", "").lower()

    if encoding == "gzip" or raw[:2] == b"\\x1f\\x8b":
        raw = gzip.decompress(raw)

    return json.loads(raw.decode("utf-8"))

records = []

for ticker, cik in COMPANIES.items():
    url = f"https://data.sec.gov/submissions/CIK{cik}.json"
    print(f"Downloading SEC submissions for {ticker} CIK {cik}")

    data = fetch_json(url)
    recent = data.get("filings", {}).get("recent", {})

    accession_numbers = recent.get("accessionNumber", [])
    forms = recent.get("form", [])
    filing_dates = recent.get("filingDate", [])
    report_dates = recent.get("reportDate", [])
    primary_docs = recent.get("primaryDocument", [])

    for i, accession in enumerate(accession_numbers[:50]):
        accession_no_dash = accession.replace("-", "")
        primary_doc = primary_docs[i] if i < len(primary_docs) else ""

        records.append({
            "ticker": ticker,
            "cik": cik,
            "company_name": data.get("name"),
            "entity_type": data.get("entityType"),
            "sic": data.get("sic"),
            "sic_description": data.get("sicDescription"),
            "owner_org": data.get("ownerOrg"),
            "state_of_incorporation": data.get("stateOfIncorporation"),
            "form_type": forms[i] if i < len(forms) else "",
            "filing_date": filing_dates[i] if i < len(filing_dates) else "",
            "report_date": report_dates[i] if i < len(report_dates) else "",
            "accession_number": accession,
            "primary_document": primary_doc,
            "filing_url": f"https://www.sec.gov/Archives/edgar/data/{int(cik)}/{accession_no_dash}/{primary_doc}",
            "source": "SEC EDGAR submissions API",
            "license": "U.S. Government public data"
        })

    time.sleep(0.2)

jsonl_path = OUT_DIR / "sec_company_submissions_sample.jsonl"
csv_path = OUT_DIR / "sec_company_submissions_sample.csv"

with jsonl_path.open("w", encoding="utf-8") as f:
    for record in records:
        f.write(json.dumps(record, ensure_ascii=False) + "\n")

with csv_path.open("w", encoding="utf-8", newline="") as f:
    writer = csv.DictWriter(f, fieldnames=list(records[0].keys()))
    writer.writeheader()
    writer.writerows(records)

print(f"Done. Records: {len(records)}")
print(f"Saved: {jsonl_path}")
print(f"Saved: {csv_path}")